diff options
author | Roger Gonzalez <roger@rogs.me> | 2023-08-17 18:27:18 +0000 |
---|---|---|
committer | Roger Gonzalez <roger@rogs.me> | 2023-08-17 18:27:18 +0000 |
commit | 286de139b7004c77185b66f1d74b17f92cb1343c (patch) | |
tree | 30414134bb431a31bad92352f8610b8381132abb | |
parent | 0cbfcae1aa1d30bdfb651b5b4404af1bf8e91a06 (diff) | |
parent | 8f6d133ff5ec4b4d5e6fdb0547984f20e1bfcf3e (diff) |
Merge branch 'refactor-to-class' into 'master'
Refactor UTE to class
See merge request rogs/ute!4
-rwxr-xr-x | pyproject.toml | 4 | ||||
-rwxr-xr-x | src/ute_wrapper/ute.py | 427 | ||||
-rw-r--r-- | src/ute_wrapper/utils.py | 85 |
3 files changed, 263 insertions, 253 deletions
diff --git a/pyproject.toml b/pyproject.toml index b6a8704..32e0b2b 100755 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,7 +45,7 @@ build-backend = "hatchling.build" [project] name = "ute_wrapper" -version = "1.0.1" +version = "1.0.2" authors = [ { name="Roger Gonzalez", email="roger@rogs.me" }, ] @@ -55,7 +55,7 @@ dependencies = [ ] readme = "README.md" requires-python = ">=3.7" -license = {file = "LICENSE"} +license = {text = "GPL version 3"} classifiers = [ "Programming Language :: Python :: 3", "Operating System :: OS Independent", diff --git a/src/ute_wrapper/ute.py b/src/ute_wrapper/ute.py index aa53bca..d41ec8a 100755 --- a/src/ute_wrapper/ute.py +++ b/src/ute_wrapper/ute.py @@ -20,209 +20,304 @@ from datetime import datetime, timedelta from time import sleep from typing import List, Optional -from .utils import make_request +import requests BASE_URL = "https://rocme.ute.com.uy/api/v1" -def login(email: str, phone_number: str) -> str: - """ - Login to UTE +class UTEClient: + def __init__(self, email: str, phone_number: str, device_id: str = None, average_cost_per_kwh: float = None): + self.email = email + self.phone_number = phone_number + self.device_id = device_id + self.average_cost_per_kwh = average_cost_per_kwh + self.authorization = self._login() + + if not self.device_id: + devices = self.get_devices_list() + + if len(devices) > 1: + devices_dict = {} + for device in devices: + devices_dict[device["name"]] = device["accountServicePointId"] + + raise Exception( + f""" + You have multiple device IDs. You need to choose one from the list + Valid options are: {devices_dict} + """ + ) + + self.device_id = devices[0]["accountServicePointId"] - Args: - email (str): User email for authentication - phone_number (str): User phone number for authentication + if not self.average_cost_per_kwh: + try: + tariff_type = self.get_account()["meterInfo"]["tariffType"].lower() + self.average_cost_per_kwh = self.get_average_price(tariff_type) + except Exception: + raise Exception("Your tariff type is not standard. Try making it explicit on the client initialization") - Returns: - str: Authorization token - """ + def _make_request(self, method: str, url: str, data: Optional[dict] = None) -> requests.Response: + """ + Make a HTTP request - url = f"{BASE_URL}/token" - data = { - "Email": email, - "PhoneNumber": phone_number, - } + Args: + method (str): The HTTP method to use. Accepted methods are ``GET``, ``POST``. + url (str): The URL to use for the request. + authorization (str): Authorization token + data (dict): The data to send in the body of the request. + + Returns: + requests.Response: The response object. - return make_request("POST", url, data=data).text + Raises: + Exception: If the method is not supported. + """ + headers = { + "X-Client-Type": "Android", + "User-Agent": "okhttp/3.8.1", + "Content-Type": "application/json; charset=utf-8", + "Connection": "Keep-Alive", + "User-Agent": "okhttp/3.8.1", + } + + try: + if self.authorization: + headers["Authorization"] = f"Bearer {self.authorization}" + except AttributeError: + pass + + if method == "GET": + return requests.get(url, headers=headers) + + if method == "POST": + return requests.post(url, headers=headers, json=data) + + raise Exception("Method not supported") + + def _login(self) -> str: + """ + Login to UTE + + Args: + email (str): User email for authentication + phone_number (str): User phone number for authentication -def get_ute_device_list(authorization: str) -> List[dict]: - """ - Get UTE device list + Returns: + str: Authorization token + """ + + url = f"{BASE_URL}/token" + data = { + "Email": self.email, + "PhoneNumber": self.phone_number, + } + + return self._make_request("POST", url, data=data).text + + def get_devices_list(self) -> List[dict]: + """ + Get UTE devices list + + Returns: + List[dict]: List of devices + """ + + accounts_url = f"{BASE_URL}/accounts" + return self._make_request("GET", accounts_url).json()["data"] + + def get_account(self) -> dict: + """ + Get UTE account info from device id + + Returns: + dict: UTE account information + """ + + accounts_by_id_url = f"{BASE_URL}/accounts/{self.device_id}" + return self._make_request("GET", accounts_by_id_url).json()["data"] + + def get_peak(self) -> dict: + """ + Get UTE peak info from device id + + Returns: + dict: UTE peak info + """ + + peak_by_id_url = f"{BASE_URL}/accounts/{self.device_id}/peak" + return self._make_request("GET", peak_by_id_url).json()["data"] + + def get_network_status(self) -> List[dict]: + """ + Get UTE network status from device id + + Returns: + dict: UTE network status + """ + + network_status_url = f"{BASE_URL}/info/network/status" + return self._make_request("GET", network_status_url).json()["data"]["summary"] + + def get_renewable_sources(self) -> str: + """ + Get UTE renewable sources + + Returns: + str: UTE renewable sources percentage + """ + + global_demand_url = f"{BASE_URL}/info/demand/global" + return self._make_request("GET", global_demand_url).json()["data"]["renewableSources"] + + def get_historic_consumption( + self, + date_start: Optional[str] = None, + date_end: Optional[str] = None, + ) -> dict: + """ + Generate UTE historic consumption from device id and date range + + Args: + date_start (str): Start date to check in format YYYY-MM-DD + date_end (str): End date to check in format YYYY-MM-DD - Returns: - List[dict]: List of devices - """ + Returns: + dict: UTE info + """ - accounts_url = f"{BASE_URL}/accounts" - return make_request("GET", accounts_url, authorization=authorization).json()["data"] + if date_start is None: + yesterday = datetime.now() - timedelta(days=1) + date_start = yesterday.strftime("%Y-%m-%d") + if date_end is None: + yesterday = datetime.now() - timedelta(days=1) + date_end = yesterday.strftime("%Y-%m-%d") -def get_ute_account_info(device_id: str, authorization: str) -> dict: - """ - Get UTE account info from device id + historic_url = ( + f"https://rocme.ute.com.uy/api/v2/device/{self.device_id}/curvefromtodate/D/{date_start}/{date_end}" + ) - Args: - device_id (str): UTE Device id + response = self._make_request("GET", historic_url).json() - Returns: - dict: UTE account info - """ + active_energy = {"total": {"sum_in_kwh": 0}} - accounts_by_id_url = f"{BASE_URL}/accounts/{device_id}" - return make_request("GET", accounts_by_id_url, authorization=authorization).json()["data"] + for item in response["data"]: + if item["magnitudeVO"] == "IMPORT_ACTIVE_ENERGY": + date = datetime.strptime(item["date"], "%Y-%m-%dT%H:%M:%S%z") + day_in_week = date.strftime("%A") + value = round(float(item["value"]), 3) + active_energy[date.strftime("%d/%m/%Y")] = { + "kwh": value, + "aproximated_cost_in_uyu": round(value * self.average_cost_per_kwh, 3), + "day_in_week": day_in_week, + } + active_energy["total"]["sum_in_kwh"] += value -def get_ute_peak_info(device_id: str, authorization: str) -> dict: - """ - Get UTE peak info from device id + active_energy["total"]["aproximated_cost_in_uyu"] = round( + active_energy["total"]["sum_in_kwh"] * self.average_cost_per_kwh, 3 + ) + active_energy["total"]["daily_average_cost"] = round( + active_energy["total"]["aproximated_cost_in_uyu"] / (len(active_energy) - 1), 3 + ) + return active_energy - Args: - device_id (str): UTE Device id + def _convert_triphasic_powers_to_power_in_watts(self, readings: List[dict]) -> float: + """ + Convert triphasic powers to power in watts - Returns: - dict: UTE peak info - """ + Args: + readings (List[dict]): List of readings - peak_by_id_url = f"{BASE_URL}/accounts/{device_id}/peak" - return make_request("GET", peak_by_id_url, authorization=authorization).json()["data"] + Returns: + float: Power in watts + """ + for reading in readings: + reading_type = reading["tipoLecturaMGMI"] + if reading_type == "I1": + i1 = float(reading["valor"]) + elif reading_type == "I2": + i2 = float(reading["valor"]) + elif reading_type == "I3": + i3 = float(reading["valor"]) + elif reading_type == "V1": + v1 = float(reading["valor"]) + elif reading_type == "V2": + v2 = float(reading["valor"]) + elif reading_type == "V3": + v3 = float(reading["valor"]) + + power_1_in_watts = v1 * i1 + power_2_in_watts = v2 * i2 + power_3_in_watts = v3 * i3 + + return round(power_1_in_watts + power_2_in_watts + power_3_in_watts, 3) -def get_ute_network_status(authorization: str) -> dict: - """ - Get UTE network status from device id + def get_current_usage_info(self) -> dict: + """ + Get current usage info from device id - Returns: - dict: UTE network status - """ + Args: + device_id (str): UTE Device id + authorization (str): Authorization token - network_status_url = f"{BASE_URL}/info/network/status" - return make_request("GET", network_status_url, authorization=authorization).json()["data"]["summary"] + Returns: + dict: UTE info + Raises: + Exception: If the reading request fails + """ -def get_ute_renewable_sources(authorization: str) -> str: - """ - Get UTE renewable sources + reading_request_url = f"{BASE_URL}/device/readingRequest" + reading_url = f"{BASE_URL}/device/{self.device_id}/lastReading/30" + + data = {"AccountServicePointId": self.device_id} - Returns: - str: UTE renewable sources percentage - """ + reading_request = self._make_request("POST", reading_request_url, data=data) - global_demand_url = f"{BASE_URL}/info/demand/global" - return make_request("GET", global_demand_url).json()["data"]["renewableSources"] + if reading_request.status_code != 200: + raise Exception("Error getting reading request") + response = self._make_request("GET", reading_url).json() -def get_ute_historic_info( - device_id: str, - authorization: str, - average_price: float, - cost_per_kwh: float, - date_start: Optional[str] = None, - date_end: Optional[str] = None, -) -> dict: - """ - Generate UTE historic info from device id and date range + while not response["success"]: + sleep(5) + response = self._make_request("GET", reading_url).json() - Args: - device_id (str): UTE Device id - authorization (str): Authorization token - cost_per_kwh (float): Cost per kwh - date_start (str): Start date to check - date_end (str): End date to check + readings = response["data"]["readings"] - Returns: - dict: UTE info - """ + power_in_watts = self._convert_triphasic_powers_to_power_in_watts(readings) - if date_start is None: - yesterday = datetime.now() - timedelta(days=1) - date_start = yesterday.strftime("%Y-%m-%d") + return_dict = {**response} + return_dict["data"]["power_in_watts"] = power_in_watts - if date_end is None: - yesterday = datetime.now() - timedelta(days=1) - date_end = yesterday.strftime("%Y-%m-%d") + return return_dict - historic_url = f"https://rocme.ute.com.uy/api/v2/device/{device_id}/curvefromtodate/D/{date_start}/{date_end}" + def get_average_price(self, plan: str) -> float: + """ + Get the average price for a plan - response = make_request("GET", historic_url, authorization=authorization).json() + Args: + plan (str): Plan name. Can be "triple" or "doble" - active_energy = {"total": {"sum_in_kwh": 0}} + Returns: + float: Average price + + Raises: + Exception: If the plan is invalid + """ + + if plan == "triple": + # 10.680 UYU/kwh * 16.67% of the day (4 hours) + # 2.223 UYU/kwh * 29.17% of the day (7 hours) + # 4.875 UYU/kwh * 54.16% of the day (13 hours) + return (10.680 * 0.1667) + (2.223 * 0.2917) + (4.875 * 0.5416) + if plan == "doble": + # 10.680 UYU/kwh * 16.67% of the day (4 hours) + # 4.280 UYU/kwh * 83.33% of the day (20 hours) + return (10.680 * 0.1667) + (4.280 * 0.8333) - for item in response["data"]: - if item["magnitudeVO"] == "IMPORT_ACTIVE_ENERGY": - date = datetime.strptime(item["date"], "%Y-%m-%dT%H:%M:%S%z") - day_in_week = date.strftime("%A") - value = round(float(item["value"]), 3) - - active_energy[date.strftime("%d/%m/%Y")] = { - "kwh": value, - "aproximated_cost_in_uyu": round(value * cost_per_kwh, 3), - "day_in_week": day_in_week, - } - active_energy["total"]["sum_in_kwh"] = active_energy["total"]["sum_in_kwh"] + value - - active_energy["total"]["aproximated_cost_in_uyu"] = round(active_energy["total"]["sum_in_kwh"] * average_price, 3) - active_energy["total"]["daily_average_cost"] = round( - active_energy["total"]["aproximated_cost_in_uyu"] / (len(active_energy) - 1), 3 - ) - return active_energy - - -def get_current_usage_info(device_id: str, authorization: str) -> dict: - """ - Get current usage info from device id - - Args: - device_id (str): UTE Device id - authorization (str): Authorization token - - Returns: - dict: UTE info - - Raises: - Exception: If the reading request fails - """ - - reading_request_url = f"{BASE_URL}/device/readingRequest" - reading_url = f"{BASE_URL}/device/{device_id}/lastReading/30" - - data = {"AccountServicePointId": device_id} - - reading_request = make_request("POST", reading_request_url, authorization=authorization, data=data) - - if reading_request.status_code != 200: - raise Exception("Error getting reading request") - - response = make_request("GET", reading_url, authorization=authorization).json() - - while not response["success"]: - sleep(5) - response = make_request("GET", reading_url, authorization=authorization).json() - - readings = response["data"]["readings"] - - for reading in readings: - reading_type = reading["tipoLecturaMGMI"] - if reading_type == "I1": - i1 = float(reading["valor"]) - elif reading_type == "I2": - i2 = float(reading["valor"]) - elif reading_type == "I3": - i3 = float(reading["valor"]) - elif reading_type == "V1": - v1 = float(reading["valor"]) - elif reading_type == "V2": - v2 = float(reading["valor"]) - elif reading_type == "V3": - v3 = float(reading["valor"]) - - power_1_in_watts = v1 * i1 - power_2_in_watts = v2 * i2 - power_3_in_watts = v3 * i3 - - power_in_watts = round(power_1_in_watts + power_2_in_watts + power_3_in_watts, 3) - - return_dict = {**response} - return_dict["data"]["power_in_watts"] = power_in_watts - - return return_dict + raise Exception("Invalid plan") diff --git a/src/ute_wrapper/utils.py b/src/ute_wrapper/utils.py deleted file mode 100644 index 530d508..0000000 --- a/src/ute_wrapper/utils.py +++ /dev/null @@ -1,85 +0,0 @@ -""" - UTE (Administración Nacional de Usinas y Trasmisiones Eléctricas) api wrapper - Copyright (C) 2023 Roger Gonzalez - - This program is free software: you can redistribute it and/or modify - it under the terms of the GNU General Public License as published by - the Free Software Foundation, either version 3 of the License, or - (at your option) any later version. - - This program is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - GNU General Public License for more details. - - You should have received a copy of the GNU General Public License - along with this program. If not, see <https://www.gnu.org/licenses/>. -""" - -from typing import Optional - -import requests - - -def make_request(method: str, url: str, authorization: str = None, data: Optional[dict] = None) -> requests.Response: - """ - Make a HTTP request - - Args: - method (str): The HTTP method to use. Accepted methods are ``GET``, ``POST``. - url (str): The URL to use for the request. - authorization (str): Authorization token - data (dict): The data to send in the body of the request. - - Returns: - requests.Response: The response object. - - Raises: - Exception: If the method is not supported. - """ - - headers = { - "X-Client-Type": "Android", - "User-Agent": "okhttp/3.8.1", - "Content-Type": "application/json; charset=utf-8", - "Connection": "Keep-Alive", - "User-Agent": "okhttp/3.8.1", - } - - if authorization: - headers["Authorization"] = f"Bearer {authorization}" - - if method == "GET": - return requests.get(url, headers=headers) - - if method == "POST": - return requests.post(url, headers=headers, json=data) - - raise Exception("Method not supported") - - -def get_average_price(plan: str) -> float: - """ - Get the average price for a plan - - Args: - plan (str): Plan name. Can be "triple" or "doble" - - Returns: - float: Average price - - Raises: - Exception: If the plan is invalid - """ - - if plan == "triple": - # 10.680 UYU/kwh * 16.67% of the day (4 hours) - # 2.223 UYU/kwh * 29.17% of the day (7 hours) - # 4.875 UYU/kwh * 54.16% of the day (13 hours) - return (10.680 * 0.1667) + (2.223 * 0.2917) + (4.875 * 0.5416) - if plan == "doble": - # 10.680 UYU/kwh * 16.67% of the day (4 hours) - # 4.280 UYU/kwh * 83.33% of the day (20 hours) - return (10.680 * 0.1667) + (4.280 * 0.8333) - - raise Exception("Invalid plan") |