Merge branch 'main' into load_save_token

src/const.py

@@ -24,10 +24,15 @@
                                'ARRAffinitySameSite=?;'
                                ' GCLB=?')
 
-GET_CONSUMER_URL = "https://iecapi.iec.co.il//api/customer"
-GET_REQUEST_READING_URL = "https://iecapi.iec.co.il//api/Consumption/RemoteReadingRange"
-GET_ELECTRIC_BILL_URL = "https://iecapi.iec.co.il//api/ElectricBillsDrawers/ElectricBills/{contract_id}/{bp_number}"
-GET_CONTRACTS_URL = "https://iecapi.iec.co.il//api/customer/contract/{bp_number}?count=1"
-GET_SINGLE_CONTRACT_URL = "https://iecapi.iec.co.il//api/customer/contract/{bp_number}?count=1"
-GET_LAST_METER_READING_URL = "https://iecapi.iec.co.il//api/Device/LastMeterReading/{contract_id}/{bp_number}"
-AUTHENTICATE_URL = "https://iecapi.iec.co.il//api/Authentication/{id}/1/-1?customErrorPage=true"
+IEC_API_BASE_URL = "https://iecapi.iec.co.il//api/"
+GET_CONSUMER_URL = IEC_API_BASE_URL + "customer"
+GET_REQUEST_READING_URL = IEC_API_BASE_URL + "Consumption/RemoteReadingRange"
+GET_ELECTRIC_BILL_URL = IEC_API_BASE_URL + "ElectricBillsDrawers/ElectricBills/{contract_id}/{bp_number}"
+GET_CONTRACTS_URL = IEC_API_BASE_URL + "customer/contract/{bp_number}?count=1"
+GET_SINGLE_CONTRACT_URL = IEC_API_BASE_URL + "customer/contract/{bp_number}?count=1"
+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/{bp_number}"
+GET_DEVICES_BY_CONTRACT_ID_URL = IEC_API_BASE_URL + "Device/{bp_number/{contract_id}"
+GET_DEVICE_TYPE_URL = IEC_API_BASE_URL + "Device/type/{bp_number}/{contract_id}/false"
+GET_BILLING_INVOICES = IEC_API_BASE_URL + "billingCollection/invoices/{bp_number}/{contract_number}"

src/data.py

@@ -4,7 +4,11 @@
 
 from src.commons import add_jwt_to_headers
 from src.const import (
+    GET_BILLING_INVOICES,
     GET_CONSUMER_URL,
+    GET_DEVICE_TYPE_URL,
+    GET_DEVICES_BY_CONTRACT_ID_URL,
+    GET_DEVICES_URL,
     GET_ELECTRIC_BILL_URL,
     GET_LAST_METER_READING_URL,
     GET_REQUEST_READING_URL,
@@ -14,7 +18,10 @@
 from src.login import IECLoginError
 from src.models.contract import GetContractResponse
 from src.models.customer import Customer
+from src.models.device import Device, Devices, GetDeviceResponse
+from src.models.device_type import DeviceType
 from src.models.electric_bill import GetElectricBillResponse
+from src.models.invoice import GetInvoicesBody, GetInvoicesResponse
 from src.models.jwt import JWT
 from src.models.meter_reading import GetLastMeterReadingResponse
 from src.models.remote_reading import RemoteReadingRequest, RemoteReadingResponse
@@ -116,3 +123,77 @@ def get_last_meter_reading(token: str, bp_number: str, contract_id: str) -> GetL
             raise IECLoginError(response.status_code, response.reason)
 
     return GetLastMeterReadingResponse.from_dict(response.json())
+
+
+def get_devices(token: str, bp_number: str) -> list[Device]:
+    """Get Device data response from IEC API."""
+    headers = add_jwt_to_headers(HEADERS_WITH_AUTH, token)
+    # sending get request and saving the response as response object
+    response = _get_url(url=GET_DEVICES_URL.format(bp_number=bp_number),
+                        headers=headers)
+
+    if response.status_code != 200:
+        print(f"Failed Login: (Code {response.status_code}): {response.reason}")
+        if len(response.content) > 0:
+            login_error_response = ErrorResponseDescriptor.from_dict(response.json())
+            raise IECLoginError(login_error_response.code, login_error_response.error)
+        else:
+            raise IECLoginError(response.status_code, response.reason)
+
+    return [Device.from_dict(device) for device in response.json()]
+
+
+def get_devices_by_contract_id(token: str, bp_number: str, contract_id: str) -> Devices:
+    """Get Device data response from IEC API."""
+    headers = add_jwt_to_headers(HEADERS_WITH_AUTH, token)
+    # sending get request and saving the response as response object
+    response = _get_url(url=GET_DEVICES_BY_CONTRACT_ID_URL.format(bp_number=bp_number, contract_id=contract_id),
+                        headers=headers)
+
+    if response.status_code != 200:
+        print(f"Failed Login: (Code {response.status_code}): {response.reason}")
+        if len(response.content) > 0:
+            login_error_response = ErrorResponseDescriptor.from_dict(response.json())
+            raise IECLoginError(login_error_response.code, login_error_response.error)
+        else:
+            raise IECLoginError(response.status_code, response.reason)
+
+    res = GetDeviceResponse.from_dict(response.json())
+    return res.data
+
+
+def get_device_type(token: str, bp_number: str, contract_id: str) -> DeviceType:
+    """Get Device Type data response from IEC API."""
+    headers = add_jwt_to_headers(HEADERS_WITH_AUTH, token)
+    # sending get request and saving the response as response object
+    response = _get_url(url=GET_DEVICE_TYPE_URL.format(bp_number=bp_number, contract_id=contract_id),
+                        headers=headers)
+
+    if response.status_code != 200:
+        print(f"Failed Login: (Code {response.status_code}): {response.reason}")
+        if len(response.content) > 0:
+            login_error_response = ErrorResponseDescriptor.from_dict(response.json())
+            raise IECLoginError(login_error_response.code, login_error_response.error)
+        else:
+            raise IECLoginError(response.status_code, response.reason)
+
+    return DeviceType.from_dict(response.json())
+
+
+def get_billing_invoices(token: str, bp_number: str, contract_id: str) -> GetInvoicesBody:
+    """Get Device Type data response from IEC API."""
+    headers = add_jwt_to_headers(HEADERS_WITH_AUTH, token)
+    # sending get request and saving the response as response object
+    response = _get_url(url=GET_BILLING_INVOICES.format(bp_number=bp_number, contract_id=contract_id),
+                        headers=headers)
+
+    if response.status_code != 200:
+        print(f"Failed Login: (Code {response.status_code}): {response.reason}")
+        if len(response.content) > 0:
+            login_error_response = ErrorResponseDescriptor.from_dict(response.json())
+            raise IECLoginError(login_error_response.code, login_error_response.error)
+        else:
+            raise IECLoginError(response.status_code, response.reason)
+
+    res = GetInvoicesResponse.from_dict(response.json())
+    return res.data

src/iec_api_client.py

@@ -5,7 +5,10 @@
 from src import data, login
 from src.models.contract import Contract
 from src.models.customer import Customer
+from src.models.device import Device, Devices
+from src.models.device_type import DeviceType
 from src.models.electric_bill import Invoices
+from src.models.invoice import GetInvoicesBody
 from src.models.jwt import JWT
 from src.models.meter_reading import MeterReadings
 from src.models.remote_reading import RemoteReadingResponse
@@ -152,6 +155,40 @@ def get_electric_bill(self, bp_number: str, contract_id: str) -> Invoices | None
             return response.data
         return None
 
+    def get_devices(self, bp_number: str) -> list[Device] | None:
+        """
+        Get a list of devices for the user
+        Args:
+            self: The instance of the class.
+            bp_number (str): The BP number of the meter.
+        Returns:
+            list[Device]: List of devices
+        """
+        self.check_token()
+
+        if not bp_number:
+            bp_number = self._bp_number
+
+        return data.get_devices(self._token, bp_number)
+
+    def get_devices_by_contract_id(self, bp_number: str, contract_id: str) -> Devices:
+        """
+        Get a list of devices for the user
+        Args:
+            self: The instance of the class.
+            bp_number (str): The BP number of the meter.
+        Returns:
+            list[Device]: List of devices
+        """
+        self.check_token()
+
+        if not contract_id:
+            contract_id = self.contract_id
+
+        if not bp_number:
+            bp_number = self._bp_number
+
+        return data.get_devices_by_contract_id(self._token, bp_number, contract_id)
 
     def get_remote_reading(self, meter_serial_number: str, meter_code: int, last_invoice_date: str, from_date: str,
                            resolution: int) -> RemoteReadingResponse:
@@ -171,6 +208,47 @@ def get_remote_reading(self, meter_serial_number: str, meter_code: int, last_inv
         return data.get_remote_reading(self._token.id_token, meter_serial_number, meter_code, last_invoice_date, from_date,
                                        resolution)
 
+
+    def get_device_type(self, bp_number: str, contract_id: str) -> DeviceType:
+        """
+        Get a list of devices for the user
+        Args:
+            self: The instance of the class.
+            bp_number (str): The BP number of the meter.
+            contract_id (str: The Contract ID
+        Returns:
+            DeviceType
+        """
+        self.check_token()
+
+        if not bp_number:
+            bp_number = self._bp_number
+
+        if not contract_id:
+            contract_id = self._contract_id
+
+        return data.get_device_type(self._token, bp_number, contract_id)
+
+    def get_billing_invoices(self, bp_number: str, contract_id: str) -> GetInvoicesBody:
+        """
+        Get a list of devices for the user
+        Args:
+            self: The instance of the class.
+            bp_number (str): The BP number of the meter.
+            contract_id (str: The Contract ID
+        Returns:
+            Billing Invoices data
+        """
+        self.check_token()
+
+        if not bp_number:
+            bp_number = self._bp_number
+
+        if not contract_id:
+            contract_id = self._contract_id
+
+        return data.get_billing_invoices(self._token, bp_number, contract_id)
+
     def check_token(self):
         """
         Check the validity of the jwt.py token and refresh in the case of expired signature errors.

src/models/device.py

@@ -0,0 +1,88 @@
+from dataclasses import dataclass, field
+
+from mashumaro import DataClassDictMixin, field_options
+
+from src.models.response_descriptor import ResponseDescriptor
+
+#
+# GET https://iecapi.iec.co.il//api/Device/{bp_number}
+#
+# [
+#     {
+#         "isActive": true,
+#         "deviceType": 3,
+#         "deviceNumber": "23231111",
+#         "deviceCode": "503"
+#     }
+#
+# -----------
+# GET https://iecapi.iec.co.il//api/Device/{bp_number}/{contract_id}
+# {
+#     "data": {
+#         "counterDevices": [
+#             {
+#                 "device": "00000000002000111",
+#                 "register": "001",
+#                 "lastMR": "00000000000011111",
+#                 "lastMRDate": "2024-01-11",
+#                 "lastMRType": "01",
+#                 "lastMRTypeDesc": "קריאת מונה לפי שגרות מערכת",
+#                 "connectionSize": {
+#                     "size": 25,
+#                     "phase": 3,
+#                     "representativeConnectionSize": "3X20"
+#                 }
+#             }
+#         ],
+#         "mrType": "01"
+#     },
+#     "reponseDescriptor": {
+#         "isSuccess": true,
+#         "code": "00",
+#         "description": null
+#     }
+# }
+
+
+@dataclass
+class Device(DataClassDictMixin):
+    """Device dataclass."""
+
+    device_type: int = field(metadata=field_options(alias="deviceType"))
+    device_number: str = field(metadata=field_options(alias="deviceNumber"))
+    device_code: str = field(metadata=field_options(alias="deviceCode"))
+    is_active: bool = field(metadata=field_options(alias="isActive"))
+
+
+@dataclass
+class ConnectionSize(DataClassDictMixin):
+    """Connection dataclass."""
+    size: int = field(metadata=field_options(alias="size"))
+    phase: str = field(metadata=field_options(alias="phase"))
+    representative_connection_size: str = field(metadata=field_options(alias="representativeConnectionSize"))
+
+
+@dataclass
+class CounterDevice(DataClassDictMixin):
+    """Counter Device dataclass."""
+    device: str = field(metadata=field_options(alias="device"))
+    register: str = field(metadata=field_options(alias="register"))
+    last_mr: str = field(metadata=field_options(alias="lastMR"))
+    last_mr_date: str = field(metadata=field_options(alias="lastMRDate"))
+    last_mr_type: str = field(metadata=field_options(alias="lastMRType"))
+    last_mr_type_desc: str = field(metadata=field_options(alias="lastMRTypeDesc"))
+    connection_size: ConnectionSize = field(metadata=field_options(alias="connectionSize"))
+
+
+@dataclass
+class Devices(DataClassDictMixin):
+    """Devices dataclass."""
+    counter_devices: list[CounterDevice]
+    mr_type: str = field(metadata=field_options(alias="mrType"))
+
+
+@dataclass
+class GetDeviceResponse(Device):
+    """Get Device Response dataclass."""
+    data: Devices
+    response_descriptor: ResponseDescriptor = field(metadata=field_options(alias="reponseDescriptor"))

src/models/device_type.py

@@ -0,0 +1,55 @@
+from dataclasses import dataclass, field
+
+from mashumaro import DataClassDictMixin, field_options
+
+from src.models.response_descriptor import ResponseDescriptor
+
+#
+# GET https://iecapi.iec.co.il//api/Device/type/{bp_number}/{contract_id}/false
+#
+# {
+#     "data": {
+#         "deviceNumber": "20008389",
+#         "deviceBalance": null,
+#         "deviceType": 0,
+#         "estimatedDaysByWeek": null,
+#         "averageUsageCostByWeek": null,
+#         "estimatedDaysByMonth": null,
+#         "averageUsageCostByMonth": null,
+#         "balanceTime": null,
+#         "balanceDate": null,
+#         "isActive": true,
+#         "numberOfDevices": 1
+#     },
+#     "reponseDescriptor": {
+#         "isSuccess": true,
+#         "code": "00",
+#         "description": ""
+#     }
+# }
+#
+
+
+@dataclass
+class DeviceType(DataClassDictMixin):
+    """Device dataclass."""
+
+    device_number: str = field(metadata=field_options(alias="deviceNumber"))
+    device_balance: int = field(metadata=field_options(alias="deviceBalance"))
+    device_type: int = field(metadata=field_options(alias="deviceType"))
+    estimated_days_by_week: int = field(metadata=field_options(alias="estimatedDaysByWeek"))
+    average_usage_cost_by_week: int = field(metadata=field_options(alias="averageUsageCostByWeek"))
+    estimated_days_by_month: int = field(metadata=field_options(alias="estimatedDaysByMonth"))
+    average_usage_cost_by_month: int = field(metadata=field_options(alias="averageUsageCostByMonth"))
+    balance_time: str = field(metadata=field_options(alias="balanceTime"))
+    balance_date: str = field(metadata=field_options(alias="balanceDate"))
+    is_active: bool = field(metadata=field_options(alias="isActive"))
+    number_of_devices: int = field(metadata=field_options(alias="numberOfDevices"))
+
+
+@dataclass
+class DeviceTypeResponse(DataClassDictMixin):
+    """Device Type Response dataclass."""
+
+    data: DeviceType
+    response_descriptor: ResponseDescriptor = field(metadata=field_options(alias="reponseDescriptor"))