Skip to content

Instantly share code, notes, and snippets.

@kleptog
Created June 12, 2018 20:03
Show Gist options
  • Save kleptog/572b529b84a1f0c40c3c69edaa18d670 to your computer and use it in GitHub Desktop.
Save kleptog/572b529b84a1f0c40c3c69edaa18d670 to your computer and use it in GitHub Desktop.
A simple example for how to export data from the Enelogic API (WSSE)
# From: https://enelogic.docs.apiary.io/#introduction/option-1.-wsse-authentication
# Go to the developer center: enelogic.com/nl/developers.
# Add your app and choose a desired redirect url (only for OAuth2 purposes, if using WSSE just fill in a URL e.g. enelogic.com)
# Fill in values below
import hashlib
import json
import os
import requests
import random
import string
from datetime import datetime, timedelta
appid = "..."
appsecret = "..."
username = "...@..."
apikey = "..."
def wsse(username, password):
nonce = ''.join(random.choice(string.letters + string.digits) for _ in range(16))
created = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
hash = hashlib.sha1(nonce + created + password).digest().encode('base64').strip()
return 'UsernameToken Username="%s", PasswordDigest="%s", Nonce="%s", Created="%s"' % (
username, hash, nonce.encode('base64').strip(), created)
def enelogic_wsse(appid, username, appsecret, apikey):
username = appid + "." + username
password = appsecret + "." + apikey
return wsse(username, password)
s = requests.Session()
s.headers.update({'Content-Type': 'application/json'})
# Get buildings
# Note: you have to regenerate the WSSE every request...
s.headers.update({'X-WSSE': enelogic_wsse(appid, username, appsecret, apikey)})
res = s.get("https://enelogic.com/api/buildings/")
res.raise_for_status()
buildings = {r['id']: r for r in res.json()}
#[
# {
# "id": 1505,
# "label": "Standaard",
# "address": "Nijverheidsweg",
# "addressNo": 25,
# "addressAdd": "A",
# "zipCode": "4529PP",
# "location": "Eede",
# "region": "Zeeland",
# "country": "NL",
# "timezone": "Europe/Amsterdam",
# "main": true
# }
#]
# Get measuringpoints
s.headers.update({'X-WSSE': enelogic_wsse(appid, username, appsecret, apikey)})
res = s.get("https://enelogic.com/api/measuringpoints/")
res.raise_for_status()
measurepoints = {r['id']: r for r in res.json()}
# {
# "id": 1651,
# "buildingId": 1505,
# "mdId": 1,
# "label": "Netbeheerder",
# "dayMin": "2013-03-14 00:00:00",
# "dayMax": "2014-02-27 23:45:00",
# "monthMin": "2013-02-15 00:00:00",
# "monthMax": "2014-02-27 00:00:00",
# "yearMin": "2013-02-01 00:00:00",
# "yearMax": "2014-02-27 00:00:00",
# "timezone": "Europe/Amsterdam"
# },
for measurepoint in measurepoints:
curr = datetime.strptime(measurepoints[measurepoint]['dayMin'][:10], "%Y-%m-%d")
end = datetime.strptime(measurepoints[measurepoint]['dayMax'][:10], "%Y-%m-%d")
while curr < end:
fname = "enelogic/" + str(measurepoint) + "." + curr.strftime("%Y-%m-%d") + '.json'
next = curr + timedelta(days=1)
if os.path.exists(fname):
curr = next
continue
print fname
s.headers.update({'X-WSSE': enelogic_wsse(appid, username, appsecret, apikey)})
res = s.get("https://enelogic.com/api/measuringpoints/%s/datapoints/%s/%s" % (measurepoint, curr.strftime("%Y-%m-%d"), next.strftime("%Y-%m-%d")))
res.raise_for_status()
with open(fname, "w") as f:
f.write(res.content)
curr = next
@edsub
Copy link

edsub commented Jan 25, 2024

Anyone that have the refresh token using OAuth2 running??

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment