feat: Add diagnostic sensors (#259)

Author: GuyKh

custom_components/iec/binary_sensor.py

@@ -20,6 +20,7 @@
     DOMAIN,
     STATICS_DICT_NAME,
     INVOICE_DICT_NAME,
+    JWT_DICT_NAME,
     EMPTY_INVOICE,
     ATTRIBUTES_DICT_NAME,
     METER_ID_ATTR_NAME,
@@ -67,7 +68,8 @@ async def async_setup_entry(
         len(
             list(
                 filter(
-                    lambda key: key != STATICS_DICT_NAME, list(coordinator.data.keys())
+                    lambda key: key not in (STATICS_DICT_NAME, JWT_DICT_NAME),
+                    list(coordinator.data.keys()),
                 )
             )
         )
@@ -76,7 +78,7 @@ async def async_setup_entry(
 
     entities: list[BinarySensorEntity] = []
     for contract_key in coordinator.data:
-        if contract_key == STATICS_DICT_NAME:
+        if contract_key in (STATICS_DICT_NAME, JWT_DICT_NAME):
             continue
 
         for description in BINARY_SENSORS:

custom_components/iec/const.py

@@ -47,6 +47,7 @@
 CONF_SELECTED_CONTRACTS = "selected_contracts"
 CONF_AVAILABLE_CONTRACTS = "contracts"
 CONF_MAIN_CONTRACT_ID = "main_contract_id"
+JWT_DICT_NAME = "jwt"
 STATICS_DICT_NAME = "statics"
 ATTRIBUTES_DICT_NAME = "entity_attributes"
 ESTIMATED_BILL_DICT_NAME = "estimated_bill"
@@ -68,3 +69,5 @@
 STATIC_KVA_TARIFF = "kva_tariff"
 STATIC_BP_NUMBER = "bp_number"
 ELECTRIC_INVOICE_DOC_ID = "1"
+ACCESS_TOKEN_ISSUED_AT = "iat"
+ACCESS_TOKEN_EXPIRATION_TIME = "exp"

custom_components/iec/coordinator.py

@@ -2,6 +2,7 @@
 
 import calendar
 import itertools
+import jwt
 import logging
 import traceback
 import socket
@@ -41,6 +42,7 @@
     DOMAIN,
     CONF_USER_ID,
     STATICS_DICT_NAME,
+    JWT_DICT_NAME,
     STATIC_KWH_TARIFF,
     INVOICE_DICT_NAME,
     FUTURE_CONSUMPTIONS_DICT_NAME,
@@ -65,6 +67,8 @@
     EST_BILL_DISTRIBUTION_PRICE_ATTR_NAME,
     EST_BILL_TOTAL_KVA_PRICE_ATTR_NAME,
     EST_BILL_KWH_CONSUMPTION_ATTR_NAME,
+    ACCESS_TOKEN_ISSUED_AT,
+    ACCESS_TOKEN_EXPIRATION_TIME,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -317,7 +321,7 @@ async def _verify_daily_readings_exist(
         )
         if not daily_reading:
             _LOGGER.debug(
-                f'Daily reading for date: {desired_date.strftime("%Y-%m-%d")} is missing, calculating manually'
+                f"Daily reading for date: {desired_date.strftime('%Y-%m-%d')} is missing, calculating manually"
             )
             readings = prefetched_reading
             if not readings:
@@ -330,7 +334,7 @@ async def _verify_daily_readings_exist(
                 )
             else:
                 _LOGGER.debug(
-                    f'Daily reading for date: {desired_date.strftime("%Y-%m-%d")} - using existing prefetched readings'
+                    f"Daily reading for date: {desired_date.strftime('%Y-%m-%d')} - using existing prefetched readings"
                 )
 
             if readings and readings.data:
@@ -353,16 +357,16 @@ async def _verify_daily_readings_exist(
                 )
                 if desired_date_reading is None or desired_date_reading.value <= 0:
                     _LOGGER.debug(
-                        f'Couldn\'t find daily reading for: {desired_date.strftime("%Y-%m-%d")}'
+                        f"Couldn't find daily reading for: {desired_date.strftime('%Y-%m-%d')}"
                     )
                 else:
                     daily_readings[device.device_number].append(
                         RemoteReading(0, desired_date, desired_date_reading.value)
                     )
         else:
             _LOGGER.debug(
-                f'Daily reading for date: {daily_reading.date.strftime("%Y-%m-%d")}'
-                f' is present: {daily_reading.value}'
+                f"Daily reading for date: {daily_reading.date.strftime('%Y-%m-%d')}"
+                f" is present: {daily_reading.value}"
             )
 
     async def _update_data(
@@ -390,12 +394,21 @@ async def _update_data(
         kwh_tariff = await self._get_kwh_tariff()
         kva_tariff = await self._get_kva_tariff()
 
+        access_token = self.api.get_token().access_token
+        decoded_token = jwt.decode(access_token, options={"verify_signature": False})
+        access_token_issued_at = decoded_token["iat"]
+        access_token_expiration_time = decoded_token["exp"]
+
         data = {
+            JWT_DICT_NAME: {
+                ACCESS_TOKEN_ISSUED_AT: access_token_issued_at,
+                ACCESS_TOKEN_EXPIRATION_TIME: access_token_expiration_time,
+            },
             STATICS_DICT_NAME: {
                 STATIC_KWH_TARIFF: kwh_tariff,
                 STATIC_KVA_TARIFF: kva_tariff,
                 STATIC_BP_NUMBER: self._bp_number,
-            }
+            },
         }
 
         estimated_bill_dict = None
@@ -454,7 +467,9 @@ async def _update_data(
 
                 devices = await self._get_devices_by_contract_id(contract_id)
                 if not devices:
-                    _LOGGER.debug(f"No devices for contract {contract_id}. Skipping creating devices.")
+                    _LOGGER.debug(
+                        f"No devices for contract {contract_id}. Skipping creating devices."
+                    )
                     continue
 
                 for device in devices or []:
@@ -705,7 +720,14 @@ async def _insert_statistics(self, contract_id: int, is_smart_meter: bool) -> No
 
                 if readings and readings.meter_start_date:
                     # Fetching the last reading from either the installation date or a month ago
-                    month_ago_time = max(month_ago_time, TIMEZONE.localize(datetime.combine(readings.meter_start_date, datetime.min.time())))
+                    month_ago_time = max(
+                        month_ago_time,
+                        TIMEZONE.localize(
+                            datetime.combine(
+                                readings.meter_start_date, datetime.min.time()
+                            )
+                        ),
+                    )
                 else:
                     _LOGGER.debug(
                         "[IEC Statistics] Failed to extract field `meterStartDate`, falling back to a month ago"
@@ -1016,8 +1038,10 @@ def _calculate_estimated_bill(
                     future_consumption_info.total_import - last_meter_read
                 )
             else:
-                _LOGGER.warn(f"Failed to calculate Future Consumption, Assuming last meter read \
-                    ({last_meter_read}) as full consumption")
+                _LOGGER.warn(
+                    f"Failed to calculate Future Consumption, Assuming last meter read \
+                    ({last_meter_read}) as full consumption"
+                )
                 future_consumption = last_meter_read
 
         kva_price = power_size * kva_tariff / 365

custom_components/iec/iec_entity.py

@@ -24,5 +24,7 @@ def __init__(
         self.meter_id = meter_id
         self.iec_entity_type = iec_entity_type
         self._attr_device_info = get_device_info(
-            self.contract_id, self.meter_id, self.iec_entity_type
+            self.contract_id if iec_entity_type != IecEntityType.GENERIC else "Generic",
+            self.meter_id,
+            self.iec_entity_type,
         )

custom_components/iec/sensor.py

@@ -5,7 +5,7 @@
 import logging
 from collections.abc import Callable
 from dataclasses import dataclass
-from datetime import datetime, timedelta, date
+from datetime import date, datetime, timedelta
 
 from homeassistant.components.sensor import (
     SensorDeviceClass,
@@ -14,36 +14,39 @@
     SensorStateClass,
 )
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import UnitOfEnergy, UnitOfTime
+from homeassistant.const import EntityCategory, UnitOfEnergy, UnitOfTime
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import StateType
 from iec_api.models.invoice import Invoice
 from iec_api.models.remote_reading import RemoteReading
 
-from .commons import find_reading_by_date, IecEntityType, TIMEZONE
+from .commons import TIMEZONE, IecEntityType, find_reading_by_date
 from .const import (
-    DOMAIN,
-    ILS,
-    STATICS_DICT_NAME,
-    STATIC_KWH_TARIFF,
-    FUTURE_CONSUMPTIONS_DICT_NAME,
-    INVOICE_DICT_NAME,
-    ILS_PER_KWH,
-    DAILY_READINGS_DICT_NAME,
-    EMPTY_REMOTE_READING,
+    ACCESS_TOKEN_EXPIRATION_TIME,
+    ACCESS_TOKEN_ISSUED_AT,
+    ATTRIBUTES_DICT_NAME,
     CONTRACT_DICT_NAME,
+    DAILY_READINGS_DICT_NAME,
+    DOMAIN,
     EMPTY_INVOICE,
-    ATTRIBUTES_DICT_NAME,
-    METER_ID_ATTR_NAME,
-    ESTIMATED_BILL_DICT_NAME,
-    TOTAL_EST_BILL_ATTR_NAME,
-    EST_BILL_DAYS_ATTR_NAME,
+    EMPTY_REMOTE_READING,
     EST_BILL_CONSUMPTION_PRICE_ATTR_NAME,
+    EST_BILL_DAYS_ATTR_NAME,
     EST_BILL_DELIVERY_PRICE_ATTR_NAME,
     EST_BILL_DISTRIBUTION_PRICE_ATTR_NAME,
-    EST_BILL_TOTAL_KVA_PRICE_ATTR_NAME,
     EST_BILL_KWH_CONSUMPTION_ATTR_NAME,
+    EST_BILL_TOTAL_KVA_PRICE_ATTR_NAME,
+    ESTIMATED_BILL_DICT_NAME,
+    FUTURE_CONSUMPTIONS_DICT_NAME,
+    ILS,
+    ILS_PER_KWH,
+    INVOICE_DICT_NAME,
+    JWT_DICT_NAME,
+    METER_ID_ATTR_NAME,
+    STATIC_KWH_TARIFF,
+    STATICS_DICT_NAME,
+    TOTAL_EST_BILL_ATTR_NAME,
 )
 from .coordinator import IecApiCoordinator
 from .iec_entity import IecEntity
@@ -120,6 +123,24 @@ def _get_reading_by_date(
         return EMPTY_REMOTE_READING
 
 
+DIAGNOSTICS_SENSORS: tuple[IecEntityDescription, ...] = (
+    IecEntityDescription(
+        key="access_token_expiry_time",
+        device_class=SensorDeviceClass.TIMESTAMP,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        value_fn=lambda data: datetime.fromtimestamp(
+            data[ACCESS_TOKEN_EXPIRATION_TIME], tz=TIMEZONE
+        ),
+    ),
+    IecEntityDescription(
+        key="access_token_issued_at",
+        device_class=SensorDeviceClass.TIMESTAMP,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        value_fn=lambda data: datetime.fromtimestamp(
+            data[ACCESS_TOKEN_ISSUED_AT], tz=TIMEZONE
+        ),
+    ),
+)
 SMART_ELEC_SENSORS: tuple[IecEntityDescription, ...] = (
     IecMeterEntityDescription(
         key="elec_forecasted_usage",
@@ -350,6 +371,16 @@ async def async_setup_entry(
                         is_multi_contract=False,
                     )
                 )
+        elif contract_key == JWT_DICT_NAME:
+            for sensor_desc in DIAGNOSTICS_SENSORS:
+                entities.append(
+                    IecSensor(
+                        coordinator,
+                        sensor_desc,
+                        JWT_DICT_NAME,
+                        is_multi_contract=False,
+                    )
+                )
         else:
             if coordinator.data[contract_key][CONTRACT_DICT_NAME].smart_meter:
                 sensors_desc: tuple[IecEntityDescription, ...] = (
@@ -425,7 +456,7 @@ def __init__(
     def native_value(self) -> StateType:
         """Return the state."""
         if self.coordinator.data is not None:
-            if self.contract_id == STATICS_DICT_NAME:
+            if self.contract_id in (STATICS_DICT_NAME, JWT_DICT_NAME):
                 return self.entity_description.value_fn(
                     self.coordinator.data.get(self.contract_id, self.meter_id)
                 )

custom_components/iec/translations/en.json

@@ -6,8 +6,11 @@
             }
         },
         "sensor": {
-            "elec_forecasted_usage": {
-                "name": "Next electric bill forecasted usage {multi_contract}"
+            "access_token_expiry_time": {
+                "name": "Access Token Expiry Time"
+            },
+            "access_token_issued_at": {
+                "name": "Access Token Issued At"
             },
             "elec_forecasted_cost": {
                 "name": "Next electric bill forecasted cost {multi_contract}",

custom_components/iec/translations/he.json

@@ -6,6 +6,12 @@
             }
         },
         "sensor": {
+            "access_token_expiry_time": {
+                "name": "תאריך פג התוקף של טוקן הגישה"
+            },
+            "access_token_issued_at": {
+                "name": "תאריך יצירת טוקן הגישה"
+            },
             "elec_forecasted_usage": {
                 "name": "סך צריכת החשמל בחשבונית הבאה {multi_contract}"
             },