Added get_tariffs to pull kWh based tariffs from Eloverblik.

pyeloverblik/__main__.py

@@ -19,6 +19,7 @@ def main():
 
     _configureLogging(args)
 
+    # Test hourly meter data.
     result = Eloverblik(args.refresh_token).get_latest(args.metering_point)
     if result.status == 200:
         total = 0
@@ -32,6 +33,7 @@ def main():
     else:
         print(f"Error getting data. Status: {result.status}. Error: {result.detailed_status}")
 
+    # Test monthly meter data.
     result = Eloverblik(args.refresh_token).get_per_month(args.metering_point)
     if result.status == 200:
         print(f"Date: {result.data_date}")
@@ -42,6 +44,15 @@ def main():
         print(f"Total: {result.get_total_metering_data()}kWh")
     else:
         print(f"Error getting data. Status: {result.status}. Error: {result.detailed_status}")
+
+    # Test fees.
+    result = Eloverblik(args.refresh_token).get_tariffs(args.metering_point)
+    if result.status == 200:
+        for fee_name, fee_rate in result.charges.items():
+            print(f"Fee '{fee_name}': {fee_rate}kr/kWh")
+
+    else:
+        print(f"Error getting data. Status: {result.status}.")
 
 def _configureLogging(args):
     if args.log:

pyeloverblik/eloverblik.py

@@ -4,11 +4,13 @@
 from datetime import datetime
 from datetime import timedelta
 import json
+from os import access
 import re
 import requests
 import logging
 from .models import RawResponse
 from .models import TimeSeries
+from .models import Charges
 from requests.adapters import HTTPAdapter
 from requests.packages.urllib3.util.retry import Retry
 
@@ -24,11 +26,14 @@
 http = requests.Session()
 http.mount("https://", adapter)
 
+
 class Eloverblik:
     '''
     Primary exported interface for eloverblik.dk API wrapper.
     '''
 
+    _access_token_cache = (None, None)
+
     def __init__(self, refresh_token):
         self._refresh_token = refresh_token
         self._base_url = 'https://api.eloverblik.dk/CustomerApi/'
@@ -73,17 +78,49 @@ def get_time_series(self,
 
         return raw_response
 
+    def get_tariffs(self,
+                    metering_point):
+        '''
+        Call charges API on eloverblik.dk and extract tariffs. Note that this does not include subscriptions or fees.
+        '''
+
+        access_token = self._get_access_token()
+        headers = self._create_headers(access_token)
+        body = '{"meteringPoints": {"meteringPoint": ["' + metering_point + '"]}}'
+        url = self._base_url + '/api/meteringpoints/meteringpoint/getcharges'
+
+        response = requests.post(url,
+                                 data=body,
+                                 headers=headers,
+                                 timeout=5
+                                 )
+
+        _LOGGER.debug(
+            f"Response from API. Status: {response.status_code}, Body: {response.text}")
+
+        if response.status_code == 200:
+            return self._parse_tariffs_from_charges_result(json.loads(response.text))
+        else:
+            return Charges(response.status_code, None, response.text)
+
     def _get_access_token(self):
+        cache_datetime, short_token = Eloverblik._access_token_cache
+
+        if cache_datetime is not None and datetime.today() - cache_datetime < timedelta(hours = 12):
+            _LOGGER.debug("Found valid token in cache.")
+            return short_token
+
         url = self._base_url + 'api/Token'
         headers = {'Authorization': 'Bearer ' + self._refresh_token}
 
         token_response = http.get(url, headers=headers, timeout=5)
         token_response.raise_for_status()
 
         token_json = token_response.json()
-
         short_token = token_json['result']
 
+        Eloverblik._access_token_cache = (datetime.today(), short_token)
+
         _LOGGER.debug(f"Got short lived token: {short_token}")
         return short_token
 
@@ -205,3 +242,27 @@ def _parse_result(self, result):
                                                f"Data most likely not available yet-3: {result}")
 
         return parsed_result
+
+    def _parse_tariffs_from_charges_result(self, result):
+        ''' 
+        Parse charges result from API call
+        '''
+
+        if 'result' in result and len(result['result']) > 0 and 'result' in result['result'][0] and 'tariffs' in result['result'][0]['result']:
+            charges = {}
+
+            for tariff in result['result'][0]['result']['tariffs']:
+                name = tariff['name'].lower().replace(' ', '_')
+
+                if tariff['periodType'] == 'P1D':
+                    charges[name] = tariff['prices'][0]['price']
+                elif tariff['periodType'] == 'PT1H':
+                    sorted_prices = [p['price'] for p in sorted(tariff['prices'], key=lambda d: d['position'])]
+                    charges[name] = sorted_prices
+                else:
+                    raise NotImplementedError(f"Unsupported periodType for tariff '{tariff['periodType']}")
+
+            return Charges(200, charges)
+
+        else:
+            return Charges(400, None);

pyeloverblik/models.py

@@ -15,12 +15,12 @@ def status(self, status):
         self._status = status
 
     @property
-    def response(self):
-        return self._response
+    def body(self):
+        return self._body
 
-    @response.setter
-    def response(self, response):
-        self._response = response
+    @body.setter
+    def body(self, response):
+        self._body = response
 
 class TimeSeries:
     '''
@@ -57,4 +57,26 @@ def get_total_metering_data(self):
         for v in self._metering_data:
             total += v
 
-        return total
+        return total
+
+class Charges:
+    '''
+    Class representing parsed charges data.
+    '''
+    def __init__(self, status, charges, detailed_status=None):
+        self._status = status
+        self._charges = charges
+        self._detailed_status = detailed_status
+
+    @property
+    def status(self):
+        return self._status
+
+    @property
+    def detailed_status(self):
+        return self._detailed_status
+
+    @property
+    def charges(self):
+        return self._charges
+