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??

@roossienb
Copy link

One of the issues is that the Enelogic server is a few seconds behind. If you use datetime.now() you may get a Forbidden, because you are doing a request in the future. Something like this works:

u = appid + "." + username
p = appsecret + "." + apikey

def create_x_wsse_header(username, password):
    nonce = uuid.uuid4().hex

    # Get current UTC time
    utc_now = datetime.datetime.now(datetime.timezone.utc)

    # Subtract 30 seconds because Enelogic server is behind on time, causing Forbidden errors.
    utc_minus_30s = utc_now - datetime.timedelta(seconds=30)
    created = utc_minus_30s.strftime("%Y-%m-%dT%H:%M:00Z")

    digest = base64.b64encode(hashlib.sha1((nonce + created + password).encode()).digest()).decode()
    base64nonce = base64.b64encode(nonce.encode("utf-8")).decode("utf-8")

    return {
        "X-WSSE": f'UsernameToken Username="{username}", PasswordDigest="{digest}", Nonce="{base64nonce}", Created="{created}"'
    }

s = requests.Session()
s.headers.update({'Content-Type': 'application/json'})    

# Get buildings

# Note: you have to regenerate the WSSE every request...
s.headers.update(create_x_wsse_header(u,p))    

@roossienb
Copy link

I've created a repository with the scripts I've developed including the above fix:
https://github.com/roossienb/enelogic-reader

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