Using the same headers for every request, improved documentation
This commit is contained in:
parent
d39453f95e
commit
8bffb19b2e
104
script.py
104
script.py
@ -1,3 +1,4 @@
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
from time import sleep
|
||||
@ -14,26 +15,33 @@ PHONE_NUMBER = os.environ.get("PHONE_NUMBER")
|
||||
|
||||
|
||||
def make_request(
|
||||
method: str, url: str, authorization: str = None, data: Optional[dict] = None, headers: dict = None
|
||||
) -> requests.Response:
|
||||
method: str, url: str, authorization: str = None, data: Optional[dict] = None) -> requests.Response:
|
||||
"""
|
||||
Make request
|
||||
Make a HTTP request
|
||||
|
||||
:param method: Method
|
||||
:param url: URL
|
||||
:param data: Data
|
||||
:return: Response
|
||||
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.
|
||||
"""
|
||||
|
||||
if not headers and authorization:
|
||||
headers = {
|
||||
"Authorization": f"Bearer {authorization}",
|
||||
"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",
|
||||
}
|
||||
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)
|
||||
@ -41,34 +49,31 @@ def make_request(
|
||||
if method == "POST":
|
||||
return requests.post(url, headers=headers, json=data)
|
||||
|
||||
return None
|
||||
raise Exception("Method not supported")
|
||||
|
||||
|
||||
def login(email: str, phone_number: str) -> requests.Response:
|
||||
def login(email: str, phone_number: str) -> str:
|
||||
"""
|
||||
Login
|
||||
Login to UTE
|
||||
|
||||
:param email: Email
|
||||
:param phone_number: Phone number
|
||||
:return: Authorization token
|
||||
Args:
|
||||
email (str): User email for authentication
|
||||
phone_number (str): User phone number for authentication
|
||||
|
||||
Returns:
|
||||
str: Authorization token
|
||||
"""
|
||||
|
||||
url = "https://rocme.ute.com.uy/api/v1/token"
|
||||
headers = {
|
||||
"X-Client-Type": "Android",
|
||||
"Content-Type": "application/json; charset=utf-8",
|
||||
"Connection": "Keep-Alive",
|
||||
"User-Agent": "okhttp/3.8.1",
|
||||
}
|
||||
data = {
|
||||
"Email": email,
|
||||
"PhoneNumber": phone_number,
|
||||
}
|
||||
|
||||
return make_request("POST", url, data=data, headers=headers).text
|
||||
return make_request("POST", url, data=data).text
|
||||
|
||||
|
||||
def generate_ute_info(
|
||||
def generate_ute_historic_info(
|
||||
device_id: str,
|
||||
authorization: str,
|
||||
cost_per_kwh: float,
|
||||
@ -76,14 +81,17 @@ def generate_ute_info(
|
||||
date_end: Optional[str] = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Generate UTE info from device id and date range
|
||||
Generate UTE historic info from device id and date range
|
||||
|
||||
:param device_id: Device id
|
||||
:param authorization: Authorization token
|
||||
:param cost_per_kwh: Cost per kwh
|
||||
:param date_start: Date start
|
||||
:param date_end: Date end
|
||||
:return: UTE info
|
||||
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
|
||||
|
||||
Returns:
|
||||
dict: UTE info
|
||||
"""
|
||||
|
||||
if date_start is None:
|
||||
@ -120,15 +128,21 @@ def generate_ute_info(
|
||||
return active_energy
|
||||
|
||||
|
||||
def get_current_usage(device_id: str, authorization: str) -> dict:
|
||||
def get_current_usage_info(device_id: str, authorization: str) -> dict:
|
||||
"""
|
||||
Get current usage
|
||||
Get current usage info from device id
|
||||
|
||||
:param device_id: Device id
|
||||
:param authorization: Authorization token
|
||||
:raises Exception: Error getting reading request
|
||||
:return: Current usage
|
||||
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 = "https://rocme.ute.com.uy/api/v1/device/readingRequest"
|
||||
reading_url = f"https://rocme.ute.com.uy/api/v1/device/{device_id}/lastReading/30"
|
||||
|
||||
@ -180,5 +194,5 @@ if __name__ == "__main__":
|
||||
average_price = (10.680 * 0.1667) + (2.223 * 0.2917) + (4.875 * 0.5416) # Each price * percentage of time on the day
|
||||
authorization = login(EMAIL, PHONE_NUMBER)
|
||||
|
||||
ute_historic_usage = generate_ute_info(DEVICE_ID, authorization, average_price, date_start, date_end)
|
||||
ute_current_usage = get_current_usage(DEVICE_ID, authorization)
|
||||
ute_historic_usage = generate_ute_historic_info(DEVICE_ID, authorization, average_price, date_start, date_end)
|
||||
ute_current_usage = get_current_usage_info(DEVICE_ID, authorization)
|
||||
|
Reference in New Issue
Block a user