diff --git a/src/const.py b/src/const.py index 06a817c..53c6c7d 100644 --- a/src/const.py +++ b/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}" diff --git a/src/data.py b/src/data.py index a9da4c7..00e90c4 100644 --- a/src/data.py +++ b/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 diff --git a/src/iec_api_client.py b/src/iec_api_client.py index a1947ac..9d0ee50 100644 --- a/src/iec_api_client.py +++ b/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. diff --git a/src/models/device.py b/src/models/device.py new file mode 100644 index 0000000..37963c3 --- /dev/null +++ b/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")) diff --git a/src/models/device_type.py b/src/models/device_type.py new file mode 100644 index 0000000..22c956d --- /dev/null +++ b/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")) diff --git a/src/models/electric_bill.py b/src/models/electric_bill.py index ce2ff36..fc65294 100644 --- a/src/models/electric_bill.py +++ b/src/models/electric_bill.py @@ -4,6 +4,7 @@ from mashumaro import DataClassDictMixin, field_options +from src.models.invoice import Invoice from src.models.response_descriptor import ResponseDescriptor # GET https://iecapi.iec.co.il//api/ElectricBillsDrawers/ElectricBills/{contract_id}/{bp_number} @@ -41,29 +42,6 @@ # } - - -@dataclass -class Invoice(DataClassDictMixin): - full_date: str = field(metadata=field_options(alias="fullDate")) - from_date: str = field(metadata=field_options(alias="fromDate")) - to_date: str = field(metadata=field_options(alias="toDate")) - amount_origin: float = field(metadata=field_options(alias="amountOrigin")) - amount_to_pay: float = field(metadata=field_options(alias="amountToPay")) - amount_paid: float = field(metadata=field_options(alias="amountPaid")) - invoice_id: int = field(metadata=field_options(alias="invoiceId")) - contract_number: int = field(metadata=field_options(alias="contractNumber")) - order_number: int = field(metadata=field_options(alias="orderNumber")) - last_date: str = field(metadata=field_options(alias="lastDate")) - invoice_payment_status: int = field( - metadata=field_options(alias="invoicePaymentStatus") - ) - document_id: str = field(metadata=field_options(alias="documentId")) - days_period: str = field(metadata=field_options(alias="daysPeriod")) - has_direct_debit: bool = field(metadata=field_options(alias="hasDirectDebit")) - invoice_type: int = field(metadata=field_options(alias="invoiceType")) - - @dataclass class Invoices(DataClassDictMixin): total_amount_to_pay: float = field(metadata=field_options(alias="totalAmountToPay")) diff --git a/src/models/invoice.py b/src/models/invoice.py new file mode 100644 index 0000000..bc9f9d1 --- /dev/null +++ b/src/models/invoice.py @@ -0,0 +1,101 @@ +from dataclasses import dataclass, field + +from mashumaro import DataClassDictMixin, field_options + +from src.models.meter_reading import MeterReading +from src.models.response_descriptor import ResponseDescriptor + +# GET https://iecapi.iec.co.il//api/billingCollection/invoices/{bp_number}/{contract_number} +# +# { +# "data": { +# "property": { +# "areaId": "111", +# "districtId": 1 +# }, +# "invoices": [ +# { +# "consumption": 123.0, +# "readingCode": 1, +# "meterReadings": [ +# { +# "reading": 123, +# "readingCode": null, +# "readingDate": "0001-01-01T00:00:00", +# "usage": null, +# "serialNumber": "000000000020008389" +# } +# ], +# "fullDate": "2020-11-01T00:00:00", +# "fromDate": "2020-09-09T00:00:00", +# "toDate": "2020-11-08T00:00:00", +# "amountOrigin": 123.51, +# "amountToPay": 0, +# "amountPaid": 123.51, +# "invoiceId": 123456, +# "contractNumber": 123456, +# "orderNumber": 0, +# "lastDate": "01/12/2020", +# "invoicePaymentStatus": 1, +# "documentID": "1", +# "daysPeriod": 61, +# "hasDirectDebit": false, +# "invoiceType": 0 +# } +# ] +# }, +# "reponseDescriptor": { +# "isSuccess": true, +# "code": "00", +# "description": "OK" +# } +# } + + +@dataclass +class Invoice(DataClassDictMixin): + full_date: str = field(metadata=field_options(alias="fullDate")) + from_date: str = field(metadata=field_options(alias="fromDate")) + to_date: str = field(metadata=field_options(alias="toDate")) + amount_origin: float = field(metadata=field_options(alias="amountOrigin")) + amount_to_pay: float = field(metadata=field_options(alias="amountToPay")) + amount_paid: float = field(metadata=field_options(alias="amountPaid")) + invoice_id: int = field(metadata=field_options(alias="invoiceId")) + contract_number: int = field(metadata=field_options(alias="contractNumber")) + order_number: int = field(metadata=field_options(alias="orderNumber")) + last_date: str = field(metadata=field_options(alias="lastDate")) + invoice_payment_status: int = field( + metadata=field_options(alias="invoicePaymentStatus") + ) + document_id: str = field(metadata=field_options(alias="documentId")) + days_period: str = field(metadata=field_options(alias="daysPeriod")) + has_direct_debit: bool = field(metadata=field_options(alias="hasDirectDebit")) + invoice_type: int = field(metadata=field_options(alias="invoiceType")) + + reading_code: int = field(metadata=field_options(alias="readingCode"), default=0) + consumption: int = field(metadata=field_options(alias="consumption"), default=0) + meter_readings: list[MeterReading] = field(metadata=field_options(alias="meterReadings"), + default_factory=lambda: []) + + +@dataclass +class Property(DataClassDictMixin): + """Property Response dataclass.""" + area_id: str = field(metadata=field_options(alias="areaId")) + district_id: int = field(metadata=field_options(alias="districtId")) + + +@dataclass +class GetInvoicesBody(DataClassDictMixin): + """Get Invoices Repnonse Response dataclass.""" + + property: Property + invoices: list[Invoice] + + +@dataclass +class GetInvoicesResponse(DataClassDictMixin): + """Device Type Response dataclass.""" + + data: GetInvoicesBody + response_descriptor: ResponseDescriptor = field(metadata=field_options(alias="reponseDescriptor"))