Dette lille Python-program, genererer CSV-filer med dit elektricitetsforbrugsdata fra Ørsted (tidligere DONG), hvis du har en fjernaflæst måler. Du kan fx bruge programmet, hvis du har lyst til at holde øje med dit forbrug i et Excel-dokument.
Det færdige program
# -*- coding: utf-8 -*-
# Author: Morten Helmstedt. E-mail: helmstedt@gmail.com
""" This program gets your Ørsted electricity consumption data and saves it to CSV"""
import requests
from datetime import datetime
from datetime import date
from datetime import timedelta
# USER ACCOUNT AND PERIOD DATA. SHOULD BE EDITED FOR YOUR NEEDS #
# User account credentials
user = '' #E-mail address
password = '' #Password
# Start date and date today used for consumption data
startdate = '2019-01-01'
today = date.today()
enddate = datetime.strftime(today, '%Y-%m-%d')
# API LOGIN #
url = 'https://api.obviux.dk/v2/authenticate'
headers = {
'X-Customer-Ip': '0.0.0.0'
}
credentials = {
'customer': user,
'password': password
}
request = requests.post(url, headers=headers, json=credentials)
response = request.json()
# Save data for further API requests
external_id = response['external_id']
headers['Authorization'] = response['token']
# GET EAN #
url = 'https://api.obviux.dk/v2/deliveries'
request = requests.get(url, headers=headers)
response = request.json()
# Assuming only one Ørsted agreement, save ean value for further API queries for that agreement
# In case of more than one agreement, loop through list and save values instead
ean = response[0]['ean']
# API CALL #
base_url = 'https://capi.obviux.dk/v1/consumption/customer/'
id_ean = external_id + '/ean/' + ean + '/'
# There's limits on periods for each type of consumption data, so we create lists of periods
startdate_datetime = datetime.strptime(startdate, '%Y-%m-%d')
enddate_datetime = datetime.strptime(enddate, '%Y-%m-%d')
# The number of days for each period
days_for_type = {
"hourly": 15,
"daily": 370,
"weekly": 420,
"monthly": 1860,
"yearly": 1830
}
def get_consumption_data(type):
# Loop that returns a list of periods (dates) to request
periods = []
loop = True
start_of_periods = startdate_datetime
days = days_for_type[type]
while loop == True:
start = datetime.strftime(start_of_periods, '%Y-%m-%d')
end = start_of_periods + timedelta(days=days)
# Loop ends and replaces end value if the calculated end date is later than what the user is looking for
if end > enddate_datetime:
end = enddate
loop = False
# Every weekly period ends on a Sunday and next starts on Monday
elif type == "weekly" and not end.weekday() == 6:
correction = end.weekday() + 1
end -= timedelta(days=correction)
end = datetime.strftime(end, '%Y-%m-%d')
start_of_periods += timedelta(days=days + 1 - correction)
# All months end on last day of month and next period starts the 1st of next month
elif type == "monthly":
day_in_month = end.day
end -= timedelta(days=day_in_month)
start_of_periods = end + timedelta(days=1)
end = datetime.strftime(end, '%Y-%m-%d')
# All yearly periods end on 31st December and next period starts next year January 1st
elif type == "yearly":
start_of_periods = datetime.strptime(str(end.year)+"-01-01", '%Y-%m-%d')
end = str(end.year-1) + "-12-31"
# Covers hourly and daily periods and weeks ending on Sunday
else:
end = datetime.strftime(end, '%Y-%m-%d')
start_of_periods += timedelta(days=days + 1)
periods.append([start, end])
# API requests to cover requested periods
url = base_url + id_ean + type
responses = []
for period in periods:
params = {
'from': period[0],
'to': period[1]
}
request = requests.get(url, headers=headers, params=params)
response = request.json()
responses.append(response)
# Write responses to CSV
if type == "hourly":
output = "date;hour;amount\n"
elif type == "weekly":
output = "date;weeknum;amount\n"
else:
output = "date;amount\n"
for entry in responses:
for data in entry['data']:
if data['consumptions']:
for d in data['consumptions']:
start = datetime.strptime(d['start'], '%Y-%m-%dT%H:%M:%S.%f%z')
start_in_timezone = datetime.astimezone(start).strftime("%Y-%m-%d %H:%M")
amount = str(d['kWh']).replace(".",",")
if type == "hourly":
hour = datetime.astimezone(start).strftime("%H")
output += start_in_timezone + ";" + hour + ";" + amount + "\n"
elif type == "weekly":
weeknum = datetime.astimezone(start).strftime("%V")
output += start_in_timezone + ";" + weeknum + ";" + amount + "\n"
else:
output += start_in_timezone + ";" + amount + "\n"
# Special case for year when change is done from manual to automatic consumption reading
if type == "yearly":
if entry['readings']:
for ent in entry['readings']:
if ent['readings']:
for e in ent['readings']:
if e['consumption']:
date = e['startdate'] + " 00:00"
amount = str(e['consumption']).replace(".",",")
output += date + ";" + amount + "\n"
filename = type + ".csv"
with open(filename, "w", encoding='utf8') as fout:
fout.write(output)
# Loop to cycle through all consumption endpoints
endpoints = ["hourly", "daily", "weekly", "monthly", "yearly"]
for endpoint in endpoints:
get_consumption_data(endpoint)