Meter

.pre-commit-config.yaml

@@ -9,7 +9,7 @@ repos:
     rev: 22.10.0
     hooks:
       - id: black
-        language_version: python3.10
+        language_version: python3.12
         args: [--line-length=100]
   - repo: https://github.com/pycqa/flake8
     rev: '5.0.4'

octopus_energy_api/account.py

@@ -5,10 +5,10 @@ def __init__(self, account_data):
             setattr(self, k, v)
 
         # setting additional values
-        #self.mpan = self.properties[0]["electricity_meter_points"][-1]["mpan"]
-        #self.serial_number = self.properties[0]["electricity_meter_points"][-1]["meters"][-1][
+        # self.mpan = self.properties[0]["electricity_meter_points"][-1]["mpan"]
+        # self.serial_number = self.properties[0]["electricity_meter_points"][-1]["meters"][-1][
         #    "serial_number"
-        #]
+        # ]
 
     def all_account_data(self):
 

octopus_energy_api/api_interface.py

@@ -1,24 +1,34 @@
 import requests
+import pandas as pd
+import time
 
 
 class api:
     def __init__(self, api_key):
         self._api_key = api_key
 
     def create_session(self):
-
         session = requests.session()
-
         session.auth = (self._api_key, "")
-
         return session
 
     def run(self, url):
-
         session = self.create_session()
-
         response = session.request(method="GET", url=url)
-
         parsed = response.json()
-
         return parsed
+
+    def pageFetcher(self, url):
+        """Recursive function to fetch all pages of results"""
+        response = self.run(url)
+        if "results" in response:
+            results = pd.DataFrame(response["results"])
+        else:
+            raise Exception(response)
+        if response["next"]:
+            # be kind to the API
+            time.sleep(5)
+            nextResults = self.pageFetcher(response["next"])
+            return pd.concat([results, nextResults], ignore_index=True)
+        else:
+            return results

octopus_energy_api/meter.py

@@ -0,0 +1,43 @@
+from datetime import datetime, timezone
+
+
+class meter:
+    def __init__(self, api, mpan, meter_data):
+        self._api = api
+        self.mpan = mpan
+        self.meter_data = meter_data
+        for k, v in meter_data.items():
+            setattr(self, k, v)
+        data = self._api.discover_meter(self)
+        if data:
+            self.hasData = True
+            self.start = data[0]
+            self.end = data[1]
+            self.count = data[2]
+        else:
+            self.hasData = False
+        print(self.printMeter())
+
+    def printMeter(self):
+        if self.hasData:
+            return (
+                "MPAN: "
+                + self.mpan
+                + " / Serial: "
+                + self.serial_number
+                + " / DataPoints: "
+                + str(self.count)
+                + " / From: "
+                + self.start
+                + " / End: "
+                + self.end
+            )
+        else:
+            return "MPAN: " + self.mpan + " / Serial: " + self.serial_number
+
+    def consumption(self, start: datetime, end: datetime):
+        """Get all consumption data between 2 datetimes"""
+        start = start.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+        end = end.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+        url = self._api._urls().consumption_url(self.mpan, self.serial_number, start, end)
+        return self._api._api.pageFetcher(url)

octopus_energy_api/meter_point.py

@@ -1,10 +1,13 @@
-class meter_point:
-    def __init__(self, urls, api, meter_point_data):
+import octopus_energy_api.meter
+
 
-        self._urls = urls
+class meter_point:
+    def __init__(self, api, meter_point_data):
         self._api = api
-
         self.meter_point_data = meter_point_data
         for k, v in meter_point_data.items():
             setattr(self, k, v)
-
+        self.m = []
+        for thismeter in self.meters:
+            am = octopus_energy_api.meter.meter(api, self.mpan, thismeter)
+            self.m.append(am)

octopus_energy_api/octopus_energy_api.py

@@ -1,8 +1,7 @@
 from octopus_energy_api.api_interface import api
 from octopus_energy_api.account import account
 from octopus_energy_api.urls import urls
-from octopus_energy_api.urls import meter_point
-
+from octopus_energy_api.meter_point import meter_point
 from datetime import datetime
 import statistics
 
@@ -20,12 +19,16 @@ def __init__(self, account_number, api_key, mpan=None, serial_number=None):
         self.account = account(account_details)
         self.properties = []
         for property in self.account.properties:
+            p = {}
             meters_points = []
-            for meter_point in property['electricity_meter_points']:
-                mp = meter_point(self._urls, self._api, meter_point)
-                meters.append(mp)
-            self.properties.append(meters_points)
-
+            for k, v in property.items():
+                if "meter_points" not in k:
+                    print(k)
+            for elec_meter_point in property["electricity_meter_points"]:
+                mp = meter_point(self, elec_meter_point)
+                meters_points.append(mp)
+            p["meters"] = meters_points
+            self.properties.append(p)
 
     def account_details(self):
         """See account data"""
@@ -36,6 +39,18 @@ def account_details(self):
 
         return response
 
+    def discover_meter(self, meter):
+        url = self._urls.meter_discovery_url(meter.mpan, meter.serial_number)
+        answer = self._api.run(url)["results"]
+        if len(answer) == 0:
+            return False
+        start = answer[0]["interval_start"]
+        url = self._urls.meter_discovery_url(meter.mpan, meter.serial_number, "-period")
+        foo = self._api.run(url)
+        end = foo["results"][0]["interval_end"]
+        count = foo["count"]
+        return [start, end, count]
+
     def products(self):
         """Get all product info for Octopus Energy"""
 
@@ -50,6 +65,13 @@ def convert_datetime_to_tz(cls, time):
 
         return time.strftime(format_tz)
 
+    @classmethod
+    def convert_to_datetime(cls, time):
+
+        format_tz = "%Y-%m-%dT%H:%M:%S%z"
+
+        return datetime.strptime(time, format_tz)
+
     def consumption(self, start: datetime, end: datetime):
         """Get all consumption data between 2 datetimes"""
 

octopus_energy_api/tariff.py

@@ -0,0 +1,60 @@
+from datetime import datetime, timezone
+import pandas as pd
+
+
+class tarrif:
+    def __init__(self, api, tariff_code, fromDT, toDT):
+        self._api = api
+        self.fuel = tariff_code[0]
+        self.registers = tariff_code[2]
+        self.productCode = tariff_code[5:-2]
+        self.GSPGroup = tariff_code[-1]
+        self.fromDT = datetime.fromisoformat(fromDT)
+        if toDT:
+            self.toDT = datetime.fromisoformat(toDT)
+        else:
+            self.toDT = datetime.now(timezone.utc)
+
+    def __str__(self):
+        return (
+            "Fuel - "
+            + self.fuel
+            + ", Registers - "
+            + self.registers
+            + ", Product Code - "
+            + self.productCode
+            + ", GSP Group - "
+            + self.GSPGroup
+        )
+
+    def tariffCode(self):
+        return self.fuel + "-" + self.registers + "R-" + self.productCode + "-" + self.GSPGroup
+
+    def lookup(self, start: datetime = None, end: datetime = None):
+        if not start:
+            start = self.fromDT
+        if not end:
+            end = self.toDT
+        if self.fuel == "E" and int(self.registers) == 1:
+            start = start.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+            end = end.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+            url = self._api._urls().tariff_url(self.productCode, self.tariffCode(), start, end)
+            dfPrice = self._api._api.pageFetcher(url)
+            dfPrice["valid_from"] = pd.to_datetime(dfPrice["valid_from"], utc=True)
+            dfPrice["valid_to"] = pd.to_datetime(dfPrice["valid_to"], utc=True)
+            # trim the end date to when we joined the tarrif
+            setend = {}
+            setend["valid_to"] = self.toDT
+            df2 = pd.DataFrame([setend])
+            dfPrice.update(df2)
+            # turn it upside down
+            dfPrice = dfPrice.sort_values("valid_from").reset_index(drop=True)
+            # trim the start date to when we joined the tarrif
+            setstart = {}
+            setstart["valid_from"] = self.fromDT
+            df2 = pd.DataFrame([setstart])
+            dfPrice.update(df2)
+
+            return dfPrice
+        else:
+            print("Only single register electric meters implemented")

octopus_energy_api/urls.py

@@ -15,16 +15,24 @@ def accounts_url(cls, account_number: str):
         return url
 
     @classmethod
-    def products_url(cls):
+    def products_url(cls, productCode: str):
+        setup = f"/v1/products/{productCode}"
 
-        url = cls.build_url("/v1/products/?brand=OCTOPUS_ENERGY")
+        url = cls.build_url(setup)
 
         return url
 
+    @classmethod
+    def tariff_url(cls, productCode: str, tariffCode: str, start, end, page_size=1500):
+        setup = f"/v1/products/{productCode}/electricity-tariffs/{tariffCode}/standard-unit-rates/"
+        params = f"?page_size={page_size}&period_from={start}&period_to={end}&order_by=period"
+        url = cls.build_url(setup, params=params)
+        return url
+
     @classmethod
     def consumption_url(cls, mpan, serial, start, end, page_size=25000):
 
-        setup = f"/v1/electricity-meter-points/{mpan}/meters/{serial}/consumption"
+        setup = f"/v1/electricity-meter-points/{mpan}/meters/{serial}/consumption/"
         params = f"?page_size={page_size}&period_from={start}&period_to={end}&order_by=period"
 
         url = cls.build_url(setup, params=params)
@@ -35,7 +43,7 @@ def consumption_url(cls, mpan, serial, start, end, page_size=25000):
     def meter_discovery_url(cls, mpan, serial, order_by="period"):
 
         setup = f"/v1/electricity-meter-points/{mpan}/meters/{serial}/consumption/"
-        params = f"?page_size=1&order_by={order_by}"
+        params = f"?period_from=2015-08-01T00:00:00Z&page_size=1&order_by={order_by}"
 
         url = cls.build_url(setup, params=params)
 

pyproject.toml

@@ -0,0 +1,18 @@
+[tool.poetry]
+name = "octopus-energy-api"
+version = "0.8.2a0"
+description = "Wrapper for communicating with the Octopus Energy API"
+authors = ["Euan Campbell <dev@euan.app>","Chris Jones <chris@brasskipper.org.uk>"]
+license = "MIT"
+readme = "README.md"
+packages = [{include = "octopus_energy_api"}]
+
+[tool.poetry.dependencies]
+python = "^3.11"
+requests = "^2.28.2"
+pytest-mock = "^3.6.1"
+pandas = "^2.1.4"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"