Octopus Agile Pico W Script

agile_rates.py

import urequests as requests
import ujson as json
import utime as time
import network
import ntptime

# Credentials to connect to Wi-Fi
ssid = '<ssid>'
password = '<ssid_password>'

def set_ntp():
    """
    Get the current date and time using an NTP server
    """

    ntptime.settime()

def get_current_time():
    """
    Get the local time in GM, gmtime is handy since this is for the UK
    Have to figure out if it needs any daylight time adjustments. Probably not
    since time is already coming from NTP server
    """

    local_time = time.gmtime()

    # Print the current GM date and time
    #print("Current date and time: {}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(local_time[0], local_time[1], local_time[2], local_time[3], local_time[4], local_time[5]))

    return local_time

def get_iso8601_datetime(datetime=None):
    """
    Format date and time in ISO8601 format
    """
    if datetime:
        now = datetime
    else:
        now = get_current_time()
    
    iso8601_format = "{:04d}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}".format(
        now[0], now[1], now[2], now[3], now[4], now[5]
    )
    return iso8601_format

def from_iso8601_format(iso_date_time):
    """
    Returns the date and time from a string formatted as ISO8601
    """
    # All times are Zulu, so remove the Z from the ISO8601 date
    datetime = iso_date_time.replace('Z','')
    # Date
    [yyyy, mm, dd] = [int(i) for i in datetime.split('T')[0].split('-')]
    # Time
    [hh, MM, ss] = [int(j) for j in datetime.split('T')[1].split(':')]
    
    return (yyyy, mm, dd, hh, MM, ss)

def is_datetime_between(datetime_check, start_datetime, end_datetime):
    """
    Compare if a given date/time tuple between two other date/time 
    """
    
    return start_datetime <= datetime_check <= end_datetime
    

def get_agile_rates():
    """
    Use the Octopus Agile API to get all the rates for the last 12 hours and the next 12 hours.
    """
    # Octopus Energy API endpoint for Agile rates
    url = "https://api.octopus.energy/v1/products/AGILE-18-02-21/electricity-tariffs/E-1R-AGILE-18-02-21-C/standard-unit-rates/"

    # Get today's date in ISO 8601 format
    # All times in Zulu/UTC
    now = get_current_time()
    # Convert from tuple to seconds so we can do math. There are 43200 seconds in an hour
    last_12 = time.mktime(now)-43200
    next_12 = time.mktime(now)+43200
    
    # Convert it back to tuple so we can format it
    last_12_date = time.localtime(last_12)
    next_12_date = time.localtime(next_12)
    start_date = get_iso8601_datetime(last_12_date)
    end_date = get_iso8601_datetime(next_12_date)
    
    # Make the API request
    response = requests.get(url + "?period_from=" + start_date + "&period_to=" + end_date)

    # Check if the request was successful
    if response.status_code == 200:
        rates = json.loads(response.text)
        return rates
    else:
        return None
    
def get_now_rate(agile_rates):
    """
    Find which rate is applicabe for a given date and time
    """
    # uPython returns day of the week and day of the year in tuple
    # drop the last 2 positions in the tuple to compare with our dates
    check_date = get_current_time()[:-2]
    #check_date = (2024, 10, 29, 23, 30, 10)
    
    for rate in agile_rates['results']:
        start_date = from_iso8601_format(rate['valid_from'])
        end_date = from_iso8601_format(rate['valid_to'])        
        if is_datetime_between(check_date, start_date, end_date):
            return rate['value_inc_vat']
        
    # if it doesn't return a rate from above, it means
    # 'now' doesn't not fall between any of the rates
    return 0
    
    

station = network.WLAN(network.STA_IF)
station.active(True)
station.connect(ssid, password)

while not station.isconnected():
    pass

print('Connected to Wi-Fi')

# Now that we are connected to Wi-Fi,
# set the correct date/time
set_ntp()

# Fetch and print today's Agile rates
rates = get_agile_rates()
print(rates)
if rates:
    print("Today's Agile rates:")
    for rate in rates["results"]:
        print("Time: {} - {}, Rate: {} p/kWh".format(rate['valid_from'], rate['valid_to'], rate['value_inc_vat']))
    print(get_now_rate(rates))
else:
    print("Failed to fetch Agile rates.")
#while True:
    # Check it has been a minute since the last iteration
    
#if rates:
    #print(get_now_rate(rates))
#else:
    # try to get rates again
    #rates = get_agile_rates(local_time)