ute/script.py

import json
import os
from datetime import datetime, timedelta
from time import sleep
from typing import Optional

import requests
from dotenv import load_dotenv

load_dotenv()

DEVICE_ID = os.environ.get("DEVICE_ID")
EMAIL = os.environ.get("EMAIL")
PHONE_NUMBER = os.environ.get("PHONE_NUMBER")


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 login(email: str, phone_number: str) -> str:
    """
    Login to UTE

    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"
    data = {
        "Email": email,
        "PhoneNumber": phone_number,
    }

    return make_request("POST", url, data=data).text


def get_ute_historic_info(
    device_id: str,
    authorization: str,
    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:
        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:
        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")

    url = f"https://rocme.ute.com.uy/api/v2/device/{device_id}/curvefromtodate/D/{date_start}/{date_end}"

    response = make_request("GET", url, authorization=authorization).json()

    active_energy = {"total": {"sum_in_kwh": 0}}

    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 = "https://rocme.ute.com.uy/api/v1/device/readingRequest"
    reading_url = f"https://rocme.ute.com.uy/api/v1/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


if __name__ == "__main__":
    date_start = "2023-08-01"
    date_end = "2023-08-31"
    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 = get_ute_historic_info(DEVICE_ID, authorization, average_price, date_start, date_end)
    ute_current_usage = get_current_usage_info(DEVICE_ID, authorization)