Add API support for gas an electricity consumption

octopus_energy/__init__.py

@@ -0,0 +1,5 @@
+"""Python client for the Octopus Energy RESTful API"""
+
+from .client import OctopusEnergyClient
+from .models import Consumption, IntervalConsumption, MeterType, UnitType
+from .exceptions import ApiError, ApiAuthenticationError

octopus_energy/client.py

@@ -0,0 +1,84 @@
+from http import HTTPStatus
+from typing import Any
+
+import requests
+from requests.auth import HTTPBasicAuth
+
+from octopus_energy.models import UnitType, Consumption, MeterType
+from octopus_energy.exceptions import ApiError, ApiAuthenticationError
+from octopus_energy.mappers import consumption_from_response
+
+_API_BASE = "https://api.octopus.energy"
+
+
+class OctopusEnergyClient:
+    """A client for interacting with the Octopus Energy RESTful API."""
+
+    def __init__(self, api_token, default_unit: UnitType = UnitType.KWH):
+        self.auth = HTTPBasicAuth(api_token, "")
+        self.default_unit = default_unit
+
+    def get_gas_consumption_v1(
+        self, mprn, serial_number, meter_type: MeterType, desired_unit_type: UnitType = UnitType.KWH
+    ) -> Consumption:
+        """Gets the consumption of gas from a specific meter.
+
+        Args:
+            mprn: The MPRN (Meter Point Reference Number) of the meter to query
+            serial_number: The serial number of the meter to query
+            meter_type: The type of the meter being queried. The octopus energy API does not tell
+                        us what the type of meter is, so we need to define this in the request
+                        to query.
+            desired_unit_type: The desired units you want the results in. This defaults to
+                               Kilowatt Hours.
+
+        Returns:
+            The consumption of gas for the meter.
+
+        """
+        return self._execute(
+            requests.get,
+            f"v1/gas-meter-points/{mprn}/meters/{serial_number}/consumption/",
+            consumption_from_response,
+            meter_type=meter_type,
+            desired_unit_type=desired_unit_type,
+        )
+
+    def get_electricity_consumption_v1(
+        self,
+        mpan: str,
+        serial_number: str,
+        meter_type: MeterType,
+        desired_unit_type: UnitType = UnitType.KWH,
+    ) -> Consumption:
+        """Gets the consumption of electricity from a specific meter.
+
+        Args:
+            mpan: The MPAN (Meter Point Administration Number) of the meter to query
+            serial_number: The serial number of the meter to query
+            meter_type: The type of the meter being queried. The octopus energy API does not tell
+                        us what the type of meter is, so we need to define this in the request to
+                        query.
+            desired_unit_type: The desired units you want the results in. This defaults to
+                               Kilowatt Hours.
+
+        Returns:
+            The consumption of gas for the meter.
+
+        """
+        return self._execute(
+            requests.get,
+            f"v1/electricity-meter-points/{mpan}/meters/{serial_number}/consumption/",
+            consumption_from_response,
+            meter_type=meter_type,
+            desired_unit_type=desired_unit_type,
+        )
+
+    def _execute(self, func, url: str, response_mapper, **kwargs) -> Any:
+        """Executes an API call to Octopus energy and maps the response."""
+        response = func(f"{_API_BASE}/{url}", auth=self.auth)
+        if not response.ok:
+            if response.status_code == HTTPStatus.UNAUTHORIZED:
+                raise ApiAuthenticationError()
+            raise ApiError("Error", response=response)
+        return response_mapper(response, **kwargs)

octopus_energy/exceptions.py

@@ -0,0 +1,15 @@
+class ApiError(Exception):
+    """An error has occurred while calling the octopus energy API"""
+
+    def __init__(self, *args: object, response) -> None:
+        super().__init__(*args)
+        self.response = response
+
+    def __str__(self) -> str:
+        return f"{self.response.status} - {self.response.text}"
+
+
+class ApiAuthenticationError(Exception):
+    """The credentials were rejected by Octopus."""
+
+    ...

octopus_energy/mappers.py

@@ -0,0 +1,58 @@
+from dateutil.parser import isoparse
+from requests import Response
+
+from octopus_energy.models import IntervalConsumption, UnitType, Consumption, MeterType
+
+_CUBIC_METERS_TO_KWH_MULTIPLIER = 11.1868
+_KWH_TO_KWH_MULTIPLIER = 1
+_UNIT_MULTIPLIERS = {
+    (UnitType.KWH.value, UnitType.CUBIC_METERS.value): (1 / _CUBIC_METERS_TO_KWH_MULTIPLIER),
+    (UnitType.CUBIC_METERS.value, UnitType.KWH.value): _CUBIC_METERS_TO_KWH_MULTIPLIER,
+}
+
+
+def consumption_from_response(
+    response: Response, meter_type: MeterType, desired_unit_type: UnitType
+) -> Consumption:
+    """Generates the Consumption model from an octopus energy API response.
+
+    Args:
+        response: The API response object.
+        meter_type: The type of meter the reading is from.
+        desired_unit_type: The desired unit for the consumption intervals. The mapping will
+                           convert from the meters units to the desired units.
+
+    Returns:
+        The Consumption model for the period of time represented in the response.
+
+    """
+    response_json = response.json()
+    if "results" not in response_json:
+        return Consumption(unit=desired_unit_type, meter_type=meter_type)
+    return Consumption(
+        desired_unit_type,
+        meter_type,
+        [
+            IntervalConsumption(
+                consumed_units=_calculate_unit(
+                    result["consumption"], meter_type.unit_type, desired_unit_type
+                ),
+                interval_start=isoparse(result["interval_start"]),
+                interval_end=isoparse(result["interval_end"]),
+            )
+            for result in response_json["results"]
+        ],
+    )
+
+
+def _calculate_unit(consumption, actual_unit, desired_unit):
+    """Converts unit values from one unit to another unit.
+
+    If no mapping is available the value is returned unchanged.
+
+    :param consumption: The consumption to convert.
+    :param actual_unit: The unit the supplied consumption is measured in.
+    :param desired_unit: The unit the convert the consumption to.
+    :return: The consumption converted to the desired unit.
+    """
+    return consumption * _UNIT_MULTIPLIERS.get((actual_unit.value, desired_unit.value), 1)

octopus_energy/models.py

@@ -0,0 +1,68 @@
+from dataclasses import dataclass, field
+from datetime import datetime
+from decimal import Decimal
+from enum import Enum
+from typing import List
+
+
+class UnitType(Enum):
+    """Units of energy measurement."""
+
+    KWH = ("kWh", "Kilowatt Hours")
+    CUBIC_METERS = ("m³", "Cubic Meters")
+
+    @property
+    def description(self) -> str:
+        """A description, in english, of the unit type."""
+        return self.value[1]
+
+    def __eq__(self, other):
+        return self.value == other.value
+
+
+class MeterType(Enum):
+    """Energy meter types, the units the measure in and description in english."""
+
+    SMETS1_GAS = ("SMETS1_GAS", UnitType.KWH, "1st Generation Smart Gas Meter")
+    SMETS2_GAS = ("SMETS2_GAS", UnitType.CUBIC_METERS, "2nd Generation Smart Gas Meter")
+    SMETS1_ELECTRICITY = (
+        "SMETS1_ELECTRICITY",
+        UnitType.KWH,
+        "1st Generation Smart Electricity Meter",
+    )
+    SMETS2_ELECTRICITY = (
+        "SMETS2_ELECTRICITY",
+        UnitType.KWH,
+        "2nd Generation Smart Electricity Meter",
+    )
+
+    @property
+    def unit_type(self) -> UnitType:
+        """The type of unit the meter measures consumption in."""
+        return self.value[1]
+
+    @property
+    def description(self) -> str:
+        """A description, in english, of the meter."""
+        return self.value[2]
+
+    def __eq__(self, other):
+        return self.value == other.value
+
+
+@dataclass
+class IntervalConsumption:
+    """Represents the consumption of energy over a single interval of time."""
+
+    interval_start: datetime
+    interval_end: datetime
+    consumed_units: Decimal
+
+
+@dataclass
+class Consumption:
+    """Consumption of energy for a list of time intervals."""
+
+    unit: UnitType
+    meter: MeterType
+    intervals: List[IntervalConsumption] = field(default_factory=lambda: [])