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)