From f3a51ad7adaf71c92cd2c446d5f3b9315399f5b8 Mon Sep 17 00:00:00 2001 From: Guy Khmelnitsky <3136012+GuyKh@users.noreply.github.com> Date: Mon, 15 Apr 2024 17:21:48 +0300 Subject: [PATCH] feat: add TenantIdentity / DeviceDetails API (#100) --- iec_api/const.py | 1 + iec_api/data.py | 27 +++++++++++++-- iec_api/iec_client.py | 38 ++++++++++++++++++--- iec_api/models/device_identity.py | 55 +++++++++++++++++++++++++++++++ 4 files changed, 114 insertions(+), 7 deletions(-) create mode 100644 iec_api/models/device_identity.py diff --git a/iec_api/const.py b/iec_api/const.py index c1bea9f..2b4dea9 100644 --- a/iec_api/const.py +++ b/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}" diff --git a/iec_api/data.py b/iec_api/data.py index 3f2c76b..2d947aa 100644 --- a/iec_api/data.py +++ b/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]: diff --git a/iec_api/iec_client.py b/iec_api/iec_client.py index 0e37eb0..de20458 100644 --- a/iec_api/iec_client.py +++ b/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: diff --git a/iec_api/models/device_identity.py b/iec_api/models/device_identity.py new file mode 100644 index 0000000..8588cfc --- /dev/null +++ b/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])