feat: add TenantIdentity / DeviceDetails API (#100)

iec_api/const.py

@@ -36,6 +36,7 @@
 GET_LAST_METER_READING_URL = IEC_API_BASE_URL + "Device/LastMeterReading/{contract_id}/{bp_number}"
 AUTHENTICATE_URL = IEC_API_BASE_URL + "Authentication/{id}/1/-1?customErrorPage=true"
 GET_DEVICES_URL = IEC_API_BASE_URL + "Device/{contract_id}"
+GET_TENANT_IDENTITY_URL = IEC_API_BASE_URL + "Tenant/Identify/{device_id}"
 GET_DEVICE_BY_DEVICE_ID_URL = GET_DEVICES_URL + "/{device_id}"
 GET_DEVICE_TYPE_URL = IEC_API_BASE_URL + "Device/type/{bp_number}/{contract_id}/false"
 GET_BILLING_INVOICES_URL = IEC_API_BASE_URL + "BillingCollection/invoices/{contract_id}/{bp_number}"

iec_api/data.py

@@ -1,7 +1,7 @@
 import base64
 import logging
 from datetime import datetime
-from typing import Optional, TypeVar
+from typing import List, Optional, TypeVar
 
 from aiohttp import ClientSession
 from mashumaro.codecs import BasicDecoder
@@ -23,6 +23,7 @@
     GET_KWH_TARIFF_URL,
     GET_LAST_METER_READING_URL,
     GET_REQUEST_READING_URL,
+    GET_TENANT_IDENTITY_URL,
     HEADERS_WITH_AUTH,
 )
 from iec_api.models.account import Account
@@ -34,6 +35,8 @@
 from iec_api.models.customer import Customer
 from iec_api.models.device import Device, Devices
 from iec_api.models.device import decoder as devices_decoder
+from iec_api.models.device_identity import DeviceDetails, DeviceIdentity
+from iec_api.models.device_identity import decoder as device_identity_decoder
 from iec_api.models.device_type import DeviceType
 from iec_api.models.device_type import decoder as device_type_decoder
 from iec_api.models.efs import EfsMessage, EfsRequestAllServices, EfsRequestSingleService
@@ -111,7 +114,7 @@ async def _post_response_with_descriptor(
     return response_with_descriptor.data
 
 
-async def get_accounts(session: ClientSession, token: JWT) -> Optional[list[Account]]:
+async def get_accounts(session: ClientSession, token: JWT) -> Optional[List[Account]]:
     """Get Accounts response from IEC API."""
     return await _get_response_with_descriptor(session, token, GET_ACCOUNTS_URL, account_decoder)
 
@@ -153,7 +156,7 @@ async def get_remote_reading(
 
 async def get_efs_messages(
     session: ClientSession, token: JWT, contract_id: str, service_code: Optional[int] = None
-) -> Optional[list[EfsMessage]]:
+) -> Optional[List[EfsMessage]]:
     """Get EFS Messages response from IEC API."""
     if service_code:
         req = EfsRequestSingleService(
@@ -223,6 +226,24 @@ async def get_devices(session: ClientSession, token: JWT, contract_id: str) -> l
     return [Device.from_dict(device) for device in response]
 
 
+async def get_device_details(session: ClientSession, token: JWT, device_id: str) -> Optional[List[DeviceDetails]]:
+    """Get Device Details response from IEC API."""
+    device_identity: DeviceIdentity = await _get_response_with_descriptor(
+        session, token, GET_TENANT_IDENTITY_URL.format(device_id=device_id), device_identity_decoder
+    )
+
+    return device_identity.device_details if device_identity else None
+
+
+async def get_device_details_by_code(
+    session: ClientSession, token: JWT, device_id: str, device_code: str
+) -> Optional[DeviceDetails]:
+    """Get Device Details response from IEC API."""
+    devices = await get_device_details(session, token, device_id)
+
+    return next((device for device in devices if device.device_code == device_code), None)
+
+
 async def get_device_by_device_id(
     session: ClientSession, token: JWT, contract_id: str, device_id: str
 ) -> Optional[Devices]:

iec_api/iec_client.py

@@ -2,7 +2,7 @@
 import atexit
 import logging
 from datetime import datetime, timedelta
-from typing import Optional
+from typing import List, Optional
 
 import aiohttp
 import jwt
@@ -13,6 +13,7 @@
 from iec_api.models.contract_check import ContractCheck
 from iec_api.models.customer import Account, Customer
 from iec_api.models.device import Device, Devices
+from iec_api.models.device_identity import DeviceDetails
 from iec_api.models.device_type import DeviceType
 from iec_api.models.efs import EfsMessage
 from iec_api.models.electric_bill import ElectricBill
@@ -89,7 +90,7 @@ async def get_customer(self) -> Optional[Customer]:
             self._bp_number = customer.bp_number
         return customer
 
-    async def get_accounts(self) -> Optional[list[Account]]:
+    async def get_accounts(self) -> Optional[List[Account]]:
         """
         Get consumer data response from IEC API.
         :return: Customer data
@@ -257,7 +258,7 @@ async def save_invoice_pdf_to_file(
             f.write(response_bytes)
             f.close()
 
-    async def get_devices(self, contract_id: Optional[str] = None) -> Optional[list[Device]]:
+    async def get_devices(self, contract_id: Optional[str] = None) -> Optional[List[Device]]:
         """
         Get a list of devices for the user
         Args:
@@ -294,6 +295,35 @@ async def get_device_by_device_id(self, device_id: str, contract_id: Optional[st
 
         return await data.get_device_by_device_id(self._session, self._token, contract_id, device_id)
 
+    async def get_device_details_by_device_id(self, device_id: str) -> Optional[List[DeviceDetails]]:
+        """
+        Get a list of devices for the user
+        Args:
+            self: The instance of the class.
+            device_id (str): The Device id.
+        Returns:
+            list[DeviceDetails]: List of device details or None
+        """
+        await self.check_token()
+
+        return await data.get_device_details(self._session, self._token, device_id)
+
+    async def get_device_details_by_device_id_and_code(
+        self, device_id: str, device_code: str
+    ) -> Optional[DeviceDetails]:
+        """
+        Get a list of devices for the user
+        Args:
+            self: The instance of the class.
+            device_id (str): The Device id.
+            device_code (str): The Device code.
+        Returns:
+            DeviceDetails: Device details or None
+        """
+        await self.check_token()
+
+        return await data.get_device_details_by_code(self._session, self._token, device_id, device_code)
+
     async def get_remote_reading(
         self,
         meter_serial_number: str,
@@ -393,7 +423,7 @@ async def get_kwh_tariff(self) -> float:
 
     async def get_efs_messages(
         self, contract_id: Optional[str] = None, service_code: Optional[int] = None
-    ) -> Optional[list[EfsMessage]]:
+    ) -> Optional[List[EfsMessage]]:
         """
         Get EFS Messages for the contract
         Args:

iec_api/models/device_identity.py

@@ -0,0 +1,55 @@
+# GET https://iecapi.iec.co.il//api/Tenant/Identify/{{device_id}}
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Optional
+
+from mashumaro import DataClassDictMixin, field_options
+from mashumaro.codecs import BasicDecoder
+
+from iec_api.models.response_descriptor import ResponseWithDescriptor
+
+# {
+#     "data": {
+#         "devicesDetails": [
+#             {
+#                 "deviceNumber": "12345",
+#                 "deviceCode": "123",
+#                 "address": "הרצל 5, חיפה",
+#                 "lastReadingValue": "12345",
+#                 "lastReadingType": "01",
+#                 "lastReadingDate": "2024-03-01T00:00:00"
+#             }
+#         ],
+#         "lastDate": "0001-01-01T00:00:00",
+#         "privateProducer": false
+#     },
+#     "reponseDescriptor": {
+#         "isSuccess": true,
+#         "code": "26",
+#         "description": "מונה אינו חד חד ערכי"
+#     }
+# }
+
+
+@dataclass
+class DeviceDetails(DataClassDictMixin):
+    """Device Details dataclass"""
+
+    device_number: str = field(metadata=field_options(alias="deviceNumber"))
+    device_code: str = field(metadata=field_options(alias="deviceCode"))
+    address: str = field(metadata=field_options(alias="address"))
+    last_reading_value: str = field(metadata=field_options(alias="lastReadingValue"))
+    last_reading_type: str = field(metadata=field_options(alias="lastReadingType"))
+    last_reading_date: datetime = field(metadata=field_options(alias="lastReadingDate"))
+
+
+@dataclass
+class DeviceIdentity(DataClassDictMixin):
+    """Devices dataclass."""
+
+    device_details: Optional[list[DeviceDetails]] = field(metadata=field_options(alias="devicesDetails"))
+    last_date: datetime = field(metadata=field_options(alias="lastDate"))
+    is_private_producer: bool = field(metadata=field_options(alias="privateProducer"))
+
+
+decoder = BasicDecoder(ResponseWithDescriptor[DeviceIdentity])