fix: get kWh tariff from usage calculator

iec_api/const.py

@@ -42,5 +42,6 @@
 GET_BILLING_INVOICES_URL = IEC_API_BASE_URL + "BillingCollection/invoices/{contract_id}/{bp_number}"
 GET_INVOICE_PDF_URL = IEC_API_BASE_URL + "BillingCollection/pdf"
 GET_KWH_TARIFF_URL = IEC_API_BASE_URL + "content/en-US/content/tariffs/contentpages/homeelectricitytariff"
+GET_CALCULATOR_GADGET_URL = IEC_API_BASE_URL + "content/en-US/calculators/gadget"
 ERROR_FIELD_NAME = "Error"
 ERROR_SUMMARY_FIELD_NAME = "errorSummary"

iec_api/data.py

@@ -1,4 +1,3 @@
-import base64
 import logging
 from datetime import datetime
 from typing import List, Optional, TypeVar
@@ -20,7 +19,6 @@
     GET_EFS_MESSAGES_URL,
     GET_ELECTRIC_BILL_URL,
     GET_INVOICE_PDF_URL,
-    GET_KWH_TARIFF_URL,
     GET_LAST_METER_READING_URL,
     GET_REQUEST_READING_URL,
     GET_TENANT_IDENTITY_URL,
@@ -288,11 +286,3 @@ async def get_invoice_pdf(
         session, url=GET_INVOICE_PDF_URL, headers=headers, json_data=request
     )
     return await response.read()
-
-
-async def get_kwh_tariff(session: ClientSession) -> float:
-    """Get Device Type data response from IEC API."""
-    response = await commons.send_get_request(session=session, url=GET_KWH_TARIFF_URL)
-    kwh_tariff_str = response["components"][1]["table"][1][2]["value"]
-
-    return float(base64.b64decode(kwh_tariff_str).decode("utf-8"))

iec_api/iec_client.py

@@ -8,7 +8,7 @@
 import jwt
 from aiohttp import ClientSession
 
-from iec_api import commons, data, login
+from iec_api import commons, data, login, static_data
 from iec_api.models.contract import Contract
 from iec_api.models.contract_check import ContractCheck
 from iec_api.models.customer import Account, Customer
@@ -22,6 +22,7 @@
 from iec_api.models.jwt import JWT
 from iec_api.models.meter_reading import MeterReadings
 from iec_api.models.remote_reading import ReadingResolution, RemoteReadingResponse
+from iec_api.usage_calculator.calculator import UsageCalculator
 
 logger = logging.getLogger(__name__)
 
@@ -417,10 +418,13 @@ async def get_kwh_tariff(self) -> float:
         if not self._kwh_tariff or self._kwh_tariff_fetch_time < datetime.now() - timedelta(hours=1):
             logger.debug("Tariff is missing or expired, fetching kwh tariff from IEC API")
             self._kwh_tariff_fetch_time = datetime.now()
-            self._kwh_tariff = await data.get_kwh_tariff(self._session)
+            self._kwh_tariff = await static_data.get_kwh_tariff(self._session)
 
         return self._kwh_tariff
 
+    async def get_usage_calculator(self) -> UsageCalculator:
+        return await static_data.get_usage_calculator(self._session)
+
     async def get_efs_messages(
         self, contract_id: Optional[str] = None, service_code: Optional[int] = None
     ) -> Optional[List[EfsMessage]]:

iec_api/static_data.py

@@ -0,0 +1,19 @@
+from aiohttp import ClientSession
+
+from iec_api.usage_calculator.calculator import UsageCalculator
+
+usage_calculator = UsageCalculator()
+
+
+async def get_usage_calculator(session: ClientSession) -> UsageCalculator:
+    """Get Usage Calculator from IEC API data."""
+
+    if not usage_calculator.is_loaded:
+        await usage_calculator.load_data(session)
+
+    return usage_calculator
+
+
+async def get_kwh_tariff(session: ClientSession) -> float:
+    calculator = await get_usage_calculator(session)
+    return calculator.get_kwh_tariff()

iec_api/usage_calculator/calculator.py

@@ -1,25 +1,50 @@
+import logging
 from datetime import timedelta
 from typing import Optional
 
+from aiohttp import ClientSession
+
+from iec_api import commons
+from iec_api.const import GET_CALCULATOR_GADGET_URL
 from iec_api.usage_calculator.consumption import Consumption
 from iec_api.usage_calculator.electric_device import ElectricDevice, PowerUnit
 from iec_api.usage_calculator.get_calculator_response import GetCalculatorResponse
 from iec_api.usage_calculator.rates import Rates
 
+logger = logging.getLogger(__name__)
+
 
 class UsageCalculator:
-    def __init__(self, get_response: dict):
-        response = GetCalculatorResponse.from_dict(get_response)
-        self.devices: list[ElectricDevice] = response.electric_devices
-        self.rates: Rates = response.gadget_calculator_rates
+    """Usage Calculator"""
+
+    def __init__(self):
+        self.devices: list[ElectricDevice] = []
+        self.rates: Rates | None = None
+        self.is_loaded = False
+
+    async def load_data(self, session: ClientSession):
+        if not self.is_loaded:
+            iec_api_data = await commons.send_get_request(session=session, url=GET_CALCULATOR_GADGET_URL)
+            response = GetCalculatorResponse.from_dict(iec_api_data)
+            self.devices: list[ElectricDevice] = response.electric_devices
+            self.rates: Rates | None = response.gadget_calculator_rates
+            self.is_loaded = True
+        else:
+            logger.info("Usage calculator data was already loaded")
 
     def get_kwh_tariff(self) -> float:
-        return self.rates.home_rate * self.rates.vat
+        if not self.is_loaded:
+            raise ValueError("Usage calculator data is not loaded")
+        return self.rates.home_rate * (1 + self.rates.vat / 100)
 
     def get_device_names(self) -> list[str]:
+        if not self.is_loaded:
+            raise ValueError("Usage calculator data is not loaded")
         return list(map(lambda device: device.name, self.devices))
 
     def get_device_info_by_name(self, name: str) -> Optional[ElectricDevice]:
+        if not self.is_loaded:
+            raise ValueError("Usage calculator data is not loaded")
         for device in self.devices:
             if device.name == name:
                 return device

iec_api/usage_calculator/electric_device.py

@@ -28,6 +28,6 @@ class ElectricDevice(DataClassDictMixin):
     calculation_resolution: CalculationResolution = field(metadata=field_options(alias="calculationResolution"))
     power: int = field(metadata=field_options(alias="power"))
     power_unit: PowerUnit = field(metadata=field_options(alias="powerUnit"))
-    average_duration_time_of_operation_in_minutes: PowerUnit = field(
+    average_duration_time_of_operation_in_minutes: float = field(
         metadata=field_options(alias="avarageDurationTimeOfOperationInMinutes")
     )