Add support for getting the tariff rate at a specific timestamp

octopus_energy/__init__.py

@@ -18,6 +18,7 @@
     ElectricityMeter,
     GasMeter,
     PageReference,
+    TariffRate,
 )
 from .exceptions import ApiAuthenticationError, ApiError, ApiNotFoundError, ApiBadRequestError
 from .rest_client import OctopusEnergyRestClient
@@ -47,4 +48,5 @@
     "ElectricityMeter",
     "GasMeter",
     "PageReference",
+    "TariffRate",
 ]

octopus_energy/client.py

@@ -1,14 +1,17 @@
-from datetime import datetime
-from typing import List
+from datetime import datetime, timedelta
+from typing import List, Optional
 
-from .mappers import meters_from_response, consumption_from_response
+from .mappers import meters_from_response, consumption_from_response, tariff_rates_from_response
 from octopus_energy import (
     Meter,
     OctopusEnergyRestClient,
     Consumption,
     EnergyType,
     SortOrder,
     PageReference,
+    EnergyTariffType,
+    RateType,
+    TariffRate,
 )
 
 
@@ -25,14 +28,12 @@ class OctopusEnergyConsumerClient:
     This client uses async i/o.
     """
 
-    def __init__(self, account_number: str, api_token: str):
+    def __init__(self, api_token: Optional[str] = None):
         """Initializes the Octopus Energy Consumer Client.
 
         Args:
-            account_number: Your Octopus Energy Account Number.
             api_token: Your Octopus Energy API Key.
         """
-        self.account_number = account_number
         self.rest_client = OctopusEnergyRestClient(api_token)
 
     def __enter__(self):
@@ -54,9 +55,13 @@ async def close(self):
         """
         await self.rest_client.close()
 
-    async def get_meters(self) -> List[Meter]:
-        """Gets all meters associated with your account."""
-        return meters_from_response(await self.rest_client.get_account_details(self.account_number))
+    async def get_meters(self, account_number: str) -> List[Meter]:
+        """Gets all meters associated with your account.
+
+        Args:
+            account_number: Your Octopus Energy Account Number.
+        """
+        return meters_from_response(await self.rest_client.get_account_details(account_number))
 
     async def get_consumption(
         self,
@@ -99,3 +104,36 @@ async def get_consumption(
 
         response = await func(meter.meter_point.id, meter.serial_number, **params)
         return consumption_from_response(response, meter)
+
+    async def get_tariff_cost(
+        self,
+        product_code: str,
+        tariff_code: str,
+        tariff_type: EnergyTariffType,
+        rate_type: RateType,
+        timestamp: datetime,
+    ) -> Optional[TariffRate]:
+        """Gets the cost of a tariff cost at a point in time.
+
+        Args:
+            product_code: The product code.
+            tariff_code: The tariff code.
+            tariff_type: The type of energy within the tariff.
+            rate_type: The type of rate.
+            timestamp: The timestamp
+
+        Returns:
+            The cost per unit of energy for the requested rate at a point in time.
+
+        """
+        response = await self.rest_client.get_tariff_v1(
+            product_code,
+            tariff_type,
+            tariff_code,
+            rate_type,
+            period_from=timestamp,
+            # Add a millisecond as the API doesn't like requests with the same start and end
+            period_to=timestamp + timedelta(seconds=1),
+        )
+        rates = tariff_rates_from_response(response)
+        return None if not rates else rates[0]

octopus_energy/mappers.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from decimal import Decimal, ROUND_DOWN
 from typing import List, Optional
 
 import dateutil
@@ -21,6 +22,7 @@
     PageReference,
     SortOrder,
     Aggregate,
+    TariffRate,
 )
 
 _CUBIC_METERS_TO_KWH_MULTIPLIER = 11.1868
@@ -29,6 +31,7 @@
     (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,
 }
+_QUANT_3DP = Decimal("0.001")
 
 
 def to_timestamp_str(timestamp: datetime) -> str:
@@ -52,6 +55,8 @@ def from_timestamp_str(timestamp: str) -> Optional[datetime]:
     Returns:
         The timestamp as a datetime object.
     """
+    if timestamp is None:
+        return None
     return dateutil.parser.isoparse(timestamp) if timestamp else None
 
 
@@ -140,6 +145,29 @@ def consumption_from_response(
     )
 
 
+def tariff_rates_from_response(response: dict) -> List[TariffRate]:
+    """Generates the list of tariff rates from an octopus energy API response.
+
+    Args:
+        response: The API response object.
+
+    Returns:
+        The List containing the rates for a specific tariff
+
+    """
+    if "results" not in response:
+        return []
+    return [
+        TariffRate(
+            Decimal(result["value_exc_vat"]).quantize(_QUANT_3DP, rounding=ROUND_DOWN),
+            Decimal(result["value_inc_vat"]).quantize(_QUANT_3DP, rounding=ROUND_DOWN),
+            from_timestamp_str(result["valid_from"]),
+            from_timestamp_str(result.get("valid_to", None)),
+        )
+        for result in response["results"]
+    ]
+
+
 def _get_page_reference(response: dict, page: str):
     if page not in response:
         return None

octopus_energy/models.py

@@ -30,6 +30,14 @@ class Tariff:
     valid_to: Optional[datetime]
 
 
+@dataclass
+class TariffRate:
+    cost_inc_vat: Decimal
+    cost_exc_vat: Decimal
+    valid_from: datetime
+    valid_to: Optional[datetime]
+
+
 class UnitType(Enum):
     """Units of energy measurement."""
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "octopus-energy"
-version = "0.1.11"
+version = "0.1.12"
 description = "Python client for the Octopus Energy RESTful API"
 authors = ["Mark Allanson <[email protected]>"]
 license = "MIT"