fork of Philip's hydro scraper for google cloud functions - based on https://gist.github.com/exterm/97d624193b6b556d259c6eb7b027d6ab

hydro_ottawa_usage.py

import functions_framework
import requests
from pycognito import Cognito
from flask import jsonify
from datetime import datetime, timedelta
from google.cloud import storage
import json
from dateutil import parser
from zoneinfo import ZoneInfo

HYDRO_USERNAME = 'your@email'
HYDRO_PASSWORD = 'your.password'

TIMEZONE = ZoneInfo("America/Toronto")
CACHE_BUCKET = 'hydro-cache'
CACHE_FILENAME = 'hourly-usage.json'

COGNITO_CLIENT_ID = '7scfcis6ecucktmp4aqi1jk6cb'
COGNITO_USER_POOL_ID = 'ca-central-1_VYnwOhMBK'

# obtained from https://hydroottawa.com/en/accounts-services/accounts/electricity-rate-selection
TIER_PRICES = {
    "Tier1": 0.093,
    "Tier2": 0.110,
}

def get_now():
    return datetime.now(TIMEZONE)

# here we do some weird stuff:
# this counter is used for home assistant, it wants a number that keeps going up
# for this, we multiply the consumption by the proportion of the passed hour.
def calculate_incrementing_counter(intervals):
    now = get_now()

    for interval in intervals:
        start = parser.isoparse(interval['startDateTime'])

        if start.hour == now.hour:
            # caveat: since these are really small numbers, we wanna make sure the total is recorded entirely
            # so we also max out 15 minutes before the end of the hour (hence 50)
            elapsed = now.minute / 45.0
            hourly_usage = interval.get("hourlyUsage", 0.0)
            return round(hourly_usage * min(max(elapsed, 0.0), 1.0), 6)

    return 0

def find_current_interval(intervals):
    current_hour = get_now().hour

    for interval in intervals:
        start = parser.isoparse(interval['startDateTime'])
        if start.hour == current_hour:
            rate_band = interval.get("rateBand")
            interval["tierPricing"] = TIER_PRICES.get(rate_band)
            interval["incrementingHourlyCounter"] = calculate_incrementing_counter(intervals)
            return interval

    return None

def get_cached_data():
    client = storage.Client()
    bucket = client.bucket(CACHE_BUCKET)
    blob = bucket.blob(CACHE_FILENAME)

    if blob.exists():
        blob.reload()
        age = get_now() - blob.updated.replace(tzinfo=TIMEZONE)
        if age < timedelta(hours=1):
            return json.loads(blob.download_as_text())
    return None

def save_cached_data(data):
    client = storage.Client()
    bucket = client.bucket(CACHE_BUCKET)
    blob = bucket.blob(CACHE_FILENAME)
    blob.upload_from_string(json.dumps(data), content_type='application/json')

def authenticate():
    user = Cognito(COGNITO_USER_POOL_ID, COGNITO_CLIENT_ID, username=HYDRO_USERNAME)
    user.authenticate(password=HYDRO_PASSWORD)
    return user.id_token, user.access_token

def get_logged_in_app_token(id_token, access_token):
    headers = {
        'x-id': id_token,
        'x-access': access_token,
        'Accept': 'application/json',
    }
    resp = requests.get('https://api-myaccount.hydroottawa.com/app-token', headers=headers)
    if resp.status_code != 200:
        raise RuntimeError(f"Failed to get app token: {resp.text}")
    return resp.headers.get('x-amzn-remapped-authorization')

def fetch_hourly_data(headers):
    usage_url = 'https://api-myaccount.hydroottawa.com/usage/consumption/hourly'
    for day_offset in range(1, 8):
        date = (get_now() - timedelta(days=day_offset)).strftime('%Y-%m-%d')
        resp = requests.post(usage_url, headers=headers, json={'date': date})

        if resp.status_code == 200:
            data = resp.json()
            if data.get("intervals"):
                return data
    return None

@functions_framework.http
def hydro_ottawa_usage(request):
    if request.path != '/casa':
        return jsonify({'error': 'Not Found'}), 404

    try:
        cached = get_cached_data()
        if cached:
            intervals = cached.get("intervals", [])
            cached["currentUsage"] = find_current_interval(intervals)
            return jsonify({'cached': True, **cached})

        id_token, access_token = authenticate()
        app_token = get_logged_in_app_token(id_token, access_token)

        headers = {
            'x-id': id_token,
            'x-access': access_token,
            'Authorization': app_token,
            'Accept': 'application/json',
        }

        hour_data = fetch_hourly_data(headers)
        if not hour_data:
            return jsonify({'error': 'No valid hourly data found in the past 7 days'}), 404

        save_cached_data(hour_data)
        intervals = hour_data.get("intervals", [])
        hour_data["currentUsage"] = find_current_interval(intervals)

        return jsonify(hour_data)

    except Exception as e:
        return jsonify({'error': str(e)}), 500