Refactor UTE to class #4
@ -20,209 +20,285 @@ from datetime import datetime, timedelta
|
|||||||
from time import sleep
|
from time import sleep
|
||||||
from typing import List, Optional
|
from typing import List, Optional
|
||||||
|
|
||||||
from .utils import make_request
|
import requests
|
||||||
|
|
||||||
BASE_URL = "https://rocme.ute.com.uy/api/v1"
|
BASE_URL = "https://rocme.ute.com.uy/api/v1"
|
||||||
|
|
||||||
|
|
||||||
def login(email: str, phone_number: str) -> str:
|
class UTEClient:
|
||||||
"""
|
def __init__(self, email: str, phone_number: str, device_id: str = None):
|
||||||
Login to UTE
|
self.email = email
|
||||||
|
self.phone_number = phone_number
|
||||||
|
self.device_id = device_id
|
||||||
|
self.authorization = self._login()
|
||||||
|
|
||||||
Args:
|
if not self.device_id:
|
||||||
email (str): User email for authentication
|
devices = self.get_devices_list()
|
||||||
phone_number (str): User phone number for authentication
|
|
||||||
|
|
||||||
Returns:
|
if len(devices) > 1:
|
||||||
str: Authorization token
|
devices_dict = {}
|
||||||
"""
|
for device in devices:
|
||||||
|
devices_dict[device["name"]] = device["accountServicePointId"]
|
||||||
|
|
||||||
url = f"{BASE_URL}/token"
|
raise Exception(
|
||||||
data = {
|
f"""
|
||||||
"Email": email,
|
You have multiple device IDs. You need to choose one from the list
|
||||||
"PhoneNumber": phone_number,
|
Valid options are: {devices_dict}
|
||||||
}
|
"""
|
||||||
|
)
|
||||||
|
|
||||||
return make_request("POST", url, data=data).text
|
self.device_id = devices[0]["accountServicePointId"]
|
||||||
|
|
||||||
|
def _make_request(self, method: str, url: str, data: Optional[dict] = None) -> requests.Response:
|
||||||
|
"""
|
||||||
|
Make a HTTP request
|
||||||
|
|
||||||
def get_ute_device_list(authorization: str) -> List[dict]:
|
Args:
|
||||||
"""
|
method (str): The HTTP method to use. Accepted methods are ``GET``, ``POST``.
|
||||||
Get UTE device list
|
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:
|
Returns:
|
||||||
List[dict]: List of devices
|
requests.Response: The response object.
|
||||||
"""
|
|
||||||
|
|
||||||
accounts_url = f"{BASE_URL}/accounts"
|
Raises:
|
||||||
return make_request("GET", accounts_url, authorization=authorization).json()["data"]
|
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",
|
||||||
|
}
|
||||||
|
|
||||||
def get_ute_account_info(device_id: str, authorization: str) -> dict:
|
try:
|
||||||
"""
|
if self.authorization:
|
||||||
Get UTE account info from device id
|
headers["Authorization"] = f"Bearer {self.authorization}"
|
||||||
|
except AttributeError:
|
||||||
|
pass
|
||||||
|
|
||||||
Args:
|
if method == "GET":
|
||||||
device_id (str): UTE Device id
|
return requests.get(url, headers=headers)
|
||||||
|
|
||||||
Returns:
|
if method == "POST":
|
||||||
dict: UTE account info
|
return requests.post(url, headers=headers, json=data)
|
||||||
"""
|
|
||||||
|
|
||||||
accounts_by_id_url = f"{BASE_URL}/accounts/{device_id}"
|
raise Exception("Method not supported")
|
||||||
return make_request("GET", accounts_by_id_url, authorization=authorization).json()["data"]
|
|
||||||
|
|
||||||
|
def _login(self) -> str:
|
||||||
|
"""
|
||||||
|
Login to UTE
|
||||||
|
|
||||||
def get_ute_peak_info(device_id: str, authorization: str) -> dict:
|
Args:
|
||||||
"""
|
email (str): User email for authentication
|
||||||
Get UTE peak info from device id
|
phone_number (str): User phone number for authentication
|
||||||
|
|
||||||
Args:
|
Returns:
|
||||||
device_id (str): UTE Device id
|
str: Authorization token
|
||||||
|
"""
|
||||||
|
|
||||||
Returns:
|
url = f"{BASE_URL}/token"
|
||||||
dict: UTE peak info
|
data = {
|
||||||
"""
|
"Email": self.email,
|
||||||
|
"PhoneNumber": self.phone_number,
|
||||||
|
}
|
||||||
|
|
||||||
peak_by_id_url = f"{BASE_URL}/accounts/{device_id}/peak"
|
return self._make_request("POST", url, data=data).text
|
||||||
return make_request("GET", peak_by_id_url, authorization=authorization).json()["data"]
|
|
||||||
|
|
||||||
|
def get_devices_list(self) -> List[dict]:
|
||||||
|
"""
|
||||||
|
Get UTE devices list
|
||||||
|
|
||||||
def get_ute_network_status(authorization: str) -> dict:
|
Returns:
|
||||||
"""
|
List[dict]: List of devices
|
||||||
Get UTE network status from device id
|
"""
|
||||||
|
|
||||||
Returns:
|
accounts_url = f"{BASE_URL}/accounts"
|
||||||
dict: UTE network status
|
return self._make_request("GET", accounts_url).json()["data"]
|
||||||
"""
|
|
||||||
|
|
||||||
network_status_url = f"{BASE_URL}/info/network/status"
|
def get_account(self) -> dict:
|
||||||
return make_request("GET", network_status_url, authorization=authorization).json()["data"]["summary"]
|
"""
|
||||||
|
Get UTE account info from device id
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: UTE account information
|
||||||
|
"""
|
||||||
|
|
||||||
def get_ute_renewable_sources(authorization: str) -> str:
|
accounts_by_id_url = f"{BASE_URL}/accounts/{self.device_id}"
|
||||||
"""
|
return self._make_request("GET", accounts_by_id_url).json()["data"]
|
||||||
Get UTE renewable sources
|
|
||||||
|
|
||||||
Returns:
|
def get_peak(self) -> dict:
|
||||||
str: UTE renewable sources percentage
|
"""
|
||||||
"""
|
Get UTE peak info from device id
|
||||||
|
|
||||||
global_demand_url = f"{BASE_URL}/info/demand/global"
|
Returns:
|
||||||
return make_request("GET", global_demand_url).json()["data"]["renewableSources"]
|
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_ute_historic_info(
|
def get_network_status(self) -> List[dict]:
|
||||||
device_id: str,
|
"""
|
||||||
authorization: str,
|
Get UTE network status from device id
|
||||||
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
|
|
||||||
|
|
||||||
Args:
|
Returns:
|
||||||
device_id (str): UTE Device id
|
dict: UTE network status
|
||||||
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
|
|
||||||
|
|
||||||
Returns:
|
network_status_url = f"{BASE_URL}/info/network/status"
|
||||||
dict: UTE info
|
return self._make_request("GET", network_status_url).json()["data"]["summary"]
|
||||||
"""
|
|
||||||
|
|
||||||
if date_start is None:
|
def get_renewable_sources(self) -> str:
|
||||||
yesterday = datetime.now() - timedelta(days=1)
|
"""
|
||||||
date_start = yesterday.strftime("%Y-%m-%d")
|
Get UTE renewable sources
|
||||||
|
|
||||||
if date_end is None:
|
Returns:
|
||||||
yesterday = datetime.now() - timedelta(days=1)
|
str: UTE renewable sources percentage
|
||||||
date_end = yesterday.strftime("%Y-%m-%d")
|
"""
|
||||||
|
|
||||||
historic_url = f"https://rocme.ute.com.uy/api/v2/device/{device_id}/curvefromtodate/D/{date_start}/{date_end}"
|
global_demand_url = f"{BASE_URL}/info/demand/global"
|
||||||
|
return self._make_request("GET", global_demand_url).json()["data"]["renewableSources"]
|
||||||
|
|
||||||
response = make_request("GET", historic_url, authorization=authorization).json()
|
def get_historic_consumption(
|
||||||
|
self,
|
||||||
|
average_cost_per_kwh: float,
|
||||||
|
date_start: Optional[str] = None,
|
||||||
|
date_end: Optional[str] = None,
|
||||||
|
) -> dict:
|
||||||
|
"""
|
||||||
|
Generate UTE historic consumption from device id and date range
|
||||||
|
|
||||||
active_energy = {"total": {"sum_in_kwh": 0}}
|
Args:
|
||||||
|
cost_per_kwh (float): Cost per kwh
|
||||||
|
date_start (str): Start date to check
|
||||||
|
date_end (str): End date to check
|
||||||
|
|
||||||
for item in response["data"]:
|
Returns:
|
||||||
if item["magnitudeVO"] == "IMPORT_ACTIVE_ENERGY":
|
dict: UTE info
|
||||||
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")] = {
|
if date_start is None:
|
||||||
"kwh": value,
|
yesterday = datetime.now() - timedelta(days=1)
|
||||||
"aproximated_cost_in_uyu": round(value * cost_per_kwh, 3),
|
date_start = yesterday.strftime("%Y-%m-%d")
|
||||||
"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)
|
if date_end is None:
|
||||||
active_energy["total"]["daily_average_cost"] = round(
|
yesterday = datetime.now() - timedelta(days=1)
|
||||||
active_energy["total"]["aproximated_cost_in_uyu"] / (len(active_energy) - 1), 3
|
date_end = yesterday.strftime("%Y-%m-%d")
|
||||||
)
|
|
||||||
return active_energy
|
|
||||||
|
|
||||||
|
historic_url = (
|
||||||
|
f"https://rocme.ute.com.uy/api/v2/device/{self.device_id}/curvefromtodate/D/{date_start}/{date_end}"
|
||||||
|
)
|
||||||
|
|
||||||
def get_current_usage_info(device_id: str, authorization: str) -> dict:
|
response = self._make_request("GET", historic_url).json()
|
||||||
"""
|
|
||||||
Get current usage info from device id
|
|
||||||
|
|
||||||
Args:
|
active_energy = {"total": {"sum_in_kwh": 0}}
|
||||||
device_id (str): UTE Device id
|
|
||||||
authorization (str): Authorization token
|
|
||||||
|
|
||||||
Returns:
|
for item in response["data"]:
|
||||||
dict: UTE info
|
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)
|
||||||
|
|
||||||
Raises:
|
active_energy[date.strftime("%d/%m/%Y")] = {
|
||||||
Exception: If the reading request fails
|
"kwh": value,
|
||||||
"""
|
"aproximated_cost_in_uyu": round(value * average_cost_per_kwh, 3),
|
||||||
|
"day_in_week": day_in_week,
|
||||||
|
}
|
||||||
|
active_energy["total"]["sum_in_kwh"] += value
|
||||||
|
|
||||||
reading_request_url = f"{BASE_URL}/device/readingRequest"
|
active_energy["total"]["aproximated_cost_in_uyu"] = round(
|
||||||
reading_url = f"{BASE_URL}/device/{device_id}/lastReading/30"
|
active_energy["total"]["sum_in_kwh"] * 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
|
||||||
|
|
||||||
data = {"AccountServicePointId": device_id}
|
def get_current_usage_info(self) -> dict:
|
||||||
|
"""
|
||||||
|
Get current usage info from device id
|
||||||
|
|
||||||
reading_request = make_request("POST", reading_request_url, authorization=authorization, data=data)
|
Args:
|
||||||
|
device_id (str): UTE Device id
|
||||||
|
authorization (str): Authorization token
|
||||||
|
|
||||||
if reading_request.status_code != 200:
|
Returns:
|
||||||
raise Exception("Error getting reading request")
|
dict: UTE info
|
||||||
|
|
||||||
response = make_request("GET", reading_url, authorization=authorization).json()
|
Raises:
|
||||||
|
Exception: If the reading request fails
|
||||||
|
"""
|
||||||
|
|
||||||
while not response["success"]:
|
reading_request_url = f"{BASE_URL}/device/readingRequest"
|
||||||
sleep(5)
|
reading_url = f"{BASE_URL}/device/{self.device_id}/lastReading/30"
|
||||||
response = make_request("GET", reading_url, authorization=authorization).json()
|
|
||||||
|
|
||||||
readings = response["data"]["readings"]
|
data = {"AccountServicePointId": self.device_id}
|
||||||
|
|
||||||
for reading in readings:
|
reading_request = self._make_request("POST", reading_request_url, data=data)
|
||||||
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
|
if reading_request.status_code != 200:
|
||||||
power_2_in_watts = v2 * i2
|
raise Exception("Error getting reading request")
|
||||||
power_3_in_watts = v3 * i3
|
|
||||||
|
|
||||||
power_in_watts = round(power_1_in_watts + power_2_in_watts + power_3_in_watts, 3)
|
response = self._make_request("GET", reading_url).json()
|
||||||
|
|
||||||
return_dict = {**response}
|
while not response["success"]:
|
||||||
return_dict["data"]["power_in_watts"] = power_in_watts
|
sleep(5)
|
||||||
|
response = self._make_request("GET", reading_url).json()
|
||||||
|
|
||||||
return return_dict
|
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
|
||||||
|
|
||||||
|
def get_average_price(self, 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")
|
||||||
|
|||||||
@ -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")
|
|
||||||
Reference in New Issue
Block a user