feat: Add Static KVA, Delivery, Distribution and PowerSize prices (#121)

iec_api/const.py

@@ -44,6 +44,7 @@
 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_PREIOD_CALCULATOR_URL = IEC_API_BASE_URL + "content/en-US/calculators/period"
 GET_CALCULATOR_GADGET_URL = IEC_API_BASE_URL + "content/en-US/calculators/gadget"
 GET_OUTAGES_URL = IEC_API_BASE_URL + "outages/transactions/{account_id}/2"
 

iec_api/iec_client.py

@@ -422,6 +422,58 @@ async def get_kwh_tariff(self) -> float:
         """Get kWh tariff"""
         return await static_data.get_kwh_tariff(self._session)
 
+    async def get_distribution_tariff(self, phase_count: Optional[int] = None) -> float:
+        """Get get_distribution tariff"""
+
+        if not phase_count:
+            devices = await self.get_devices()
+
+            assert devices, "No Devices found"
+            device = devices[0]
+
+            device_details = await self.get_device_by_device_id(device.device_number)
+            assert device_details, "No Device Details"
+
+            phase_count = device_details.counter_devices[0].connection_size.phase
+
+        return await static_data.get_distribution_tariff(self._session, phase_count)
+
+    async def get_delivery_tariff(self, phase_count: Optional[int] = None) -> float:
+        """Get delivery tariff"""
+
+        if not phase_count:
+            devices = await self.get_devices()
+
+            assert devices, "No Devices found"
+            device = devices[0]
+
+            device_details = await self.get_device_by_device_id(device.device_number)
+            assert device_details, "No Device Details"
+
+            phase_count = device_details.counter_devices[0].connection_size.phase
+
+        return await static_data.get_delivery_tariff(self._session, phase_count)
+
+    async def get_kva_tariff(self) -> float:
+        """Get KVA tariff"""
+        return await static_data.get_kva_tariff(self._session)
+
+    async def get_power_size(self, connection: Optional[str] = None) -> float:
+        """Get delivery tariff"""
+
+        if not connection:
+            devices = await self.get_devices()
+
+            assert devices, "No Devices found"
+            device = devices[0]
+
+            device_details = await self.get_device_by_device_id(device.device_number)
+            assert device_details, "No Device Details"
+
+            connection = device_details.counter_devices[0].connection_size.representative_connection_size
+
+        return await static_data.get_power_size(self._session, connection)
+
     async def get_usage_calculator(self) -> UsageCalculator:
         """
         Get Usage Calculator module
@@ -484,7 +536,7 @@ async def get_fault_portal_user_profile(self) -> Optional[UserProfile]:
 
     async def get_fault_portal_outages_by_account(
         self, account_id: Optional[str] = None
-    ) -> (Optional)[List[FaultPortalOutage]]:
+    ) -> Optional[List[FaultPortalOutage]]:
         """Get Outages for the Account from Fault Portal
         Args:
             self: The instance of the class.

iec_api/models/device.py

@@ -62,7 +62,7 @@ class ConnectionSize(DataClassDictMixin):
     """Connection dataclass."""
 
     size: int = field(metadata=field_options(alias="size"))
-    phase: str = field(metadata=field_options(alias="phase"))
+    phase: int = field(metadata=field_options(alias="phase"))
     representative_connection_size: str = field(metadata=field_options(alias="representativeConnectionSize"))
 
 

iec_api/static_data.py

@@ -1,8 +1,22 @@
+import base64
+from decimal import Decimal
+
 from aiohttp import ClientSession
 
+from iec_api import commons
+from iec_api.const import GET_KWH_TARIFF_URL, GET_PREIOD_CALCULATOR_URL
 from iec_api.usage_calculator.calculator import UsageCalculator
 
 usage_calculator = UsageCalculator()
+cache = {}
+distribution_1p_tariff_key = "distribution_1p_tariff"
+distribution_3p_tariff_key = "distribution_3p_tariff"
+delivery_1p_tariff_key = "delivery_1p_tariff"
+delivery_3p_tariff_key = "delivery_3p_tariff"
+kwh_tariff_key = "kwh_tariff"
+kva_tariff_key = "kva_tariff"
+connection_to_power_size_key = "connection_to_power_size"
+vat_key = "vat"
 
 
 async def get_usage_calculator(session: ClientSession) -> UsageCalculator:
@@ -15,5 +29,101 @@ async def get_usage_calculator(session: ClientSession) -> UsageCalculator:
 
 
 async def get_kwh_tariff(session: ClientSession) -> float:
-    calculator = await get_usage_calculator(session)
-    return calculator.get_kwh_tariff()
+    if kwh_tariff_key not in cache:
+        calculator = await get_usage_calculator(session)
+        kwh_tariff = calculator.get_kwh_tariff()
+        cache[kwh_tariff_key] = kwh_tariff
+
+    return cache[kwh_tariff_key]
+
+
+async def _get_tariffs(session: ClientSession) -> tuple[float, float, float, float, 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"]
+    kwh_tariff = float(base64.b64decode(kwh_tariff_str).decode("utf-8"))
+
+    distribution_1p_tariff_str = response["components"][2]["table"][1][2]["value"]
+    distribution_1p_tariff = float(base64.b64decode(distribution_1p_tariff_str).decode("utf-8"))
+    distribution_3p_tariff_str = response["components"][2]["table"][2][2]["value"]
+    distribution_3p_tariff = float(base64.b64decode(distribution_3p_tariff_str).decode("utf-8"))
+
+    delivery_1p_tariff_str = response["components"][3]["table"][1][2]["value"]
+    delivery_3p_tariff_str = response["components"][3]["table"][2][2]["value"]
+    delivery_1p_tariff = float(base64.b64decode(delivery_1p_tariff_str).decode("utf-8"))
+    delivery_3p_tariff = float(base64.b64decode(delivery_3p_tariff_str).decode("utf-8"))
+
+    kva_tariff_str = response["components"][5]["table"][1][2]["value"]
+    kva_tariff = float(base64.b64decode(kva_tariff_str).decode("utf-8"))
+
+    cache[distribution_1p_tariff_key] = distribution_1p_tariff
+    cache[distribution_3p_tariff_key] = distribution_3p_tariff
+    cache[delivery_1p_tariff_key] = delivery_1p_tariff
+    cache[delivery_3p_tariff_key] = delivery_3p_tariff
+    cache[kva_tariff_key] = kva_tariff
+
+    return kwh_tariff, distribution_1p_tariff, distribution_3p_tariff, delivery_1p_tariff, delivery_3p_tariff
+
+
+async def get_distribution_tariff(session: ClientSession, phase_count: int) -> float:
+    """Get distribution tariff (incl. VAT) from IEC API."""
+
+    key = distribution_3p_tariff_key if phase_count == 3 else distribution_1p_tariff_key
+    if key not in cache:
+        await _get_tariffs(session)
+
+    return cache[key]
+
+
+async def get_delivery_tariff(session: ClientSession, phase_count: int) -> float:
+    """Get delivery tariff (incl. VAT) from IEC API."""
+
+    key = delivery_3p_tariff_key if phase_count == 3 else delivery_1p_tariff_key
+    if key not in cache:
+        await _get_tariffs(session)
+
+    return cache[key]
+
+
+async def get_kva_tariff(session: ClientSession) -> float:
+    """Get KVA tariff (incl. VAT) from IEC API."""
+
+    key = kva_tariff_key
+    if key not in cache:
+        await _get_tariffs(session)
+
+    return cache[key]
+
+
+async def _get_vat(session: ClientSession) -> Decimal:
+    """Get VAT from IEC API."""
+
+    key = vat_key
+    if key not in cache:
+        calculator = await get_usage_calculator(session)
+        cache[key] = calculator.get_vat()
+
+    return cache[key]
+
+
+async def _get_connection_to_power_size(session: ClientSession) -> dict[str, float]:
+    """Get Device Type data response from IEC API."""
+    resp = await commons.send_get_request(session=session, url=GET_PREIOD_CALCULATOR_URL)
+    connection_to_power_size_map = resp["period_Calculator_Rates"]["connectionToPowerSize"]
+
+    cache[connection_to_power_size_key] = connection_to_power_size_map
+    return connection_to_power_size_map
+
+
+async def get_power_size(session: ClientSession, connection: str) -> float:
+    """Get PowerSize by Connection (incl. VAT) from IEC API."""
+
+    key = connection_to_power_size_key
+    if key not in cache:
+        await _get_connection_to_power_size(session)
+
+    connection_to_power_size_map = cache[key]
+    power_size = connection_to_power_size_map[connection]
+
+    vat = await _get_vat(session)
+    return round(power_size * (1 + float(vat)), 2)

iec_api/usage_calculator/calculator.py

@@ -33,6 +33,11 @@ async def load_data(self, session: ClientSession):
         else:
             logger.info("Usage calculator data was already loaded")
 
+    def get_vat(self) -> float:
+        if not self.is_loaded:
+            raise ValueError("Usage calculator data is not loaded")
+        return self.rates.vat / 100
+
     def get_kwh_tariff(self) -> float:
         if not self.is_loaded:
             raise ValueError("Usage calculator data is not loaded")

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "iec-api"
-version = "0.3.0"
+version = "0.3.1"
 description = "A Python wrapper for Israel Electric Company API"
 authors = ["GuyKh"]
 license = "MIT"