Merge branch 'refactor-to-class' into 'master'

Refactor UTE to class

See merge request rogs/ute!4

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",

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:
+class UTEClient:
-    """
+    def __init__(self, email: str, phone_number: str, device_id: str = None, average_cost_per_kwh: float = None):
-    Login to UTE
+        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()
 
-    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"]
 
+        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")
 
-def get_ute_device_list(authorization: str) -> List[dict]:
+    def _make_request(self, method: str, url: str, data: Optional[dict] = None) -> requests.Response:
-    """
+        """
-    Get UTE device list
+        Make a HTTP request
 
-    Returns:
+        Args:
-        List[dict]: List of devices
+            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.
 
-    accounts_url = f"{BASE_URL}/accounts"
+        Returns:
-    return make_request("GET", accounts_url, authorization=authorization).json()["data"]
+            requests.Response: The response object.
 
+        Raises:
+            Exception: If the method is not supported.
+        """
 
-def get_ute_account_info(device_id: str, authorization: str) -> dict:
+        headers = {
-    """
+            "X-Client-Type": "Android",
-    Get UTE account info from device id
+            "User-Agent": "okhttp/3.8.1",
+            "Content-Type": "application/json; charset=utf-8",
+            "Connection": "Keep-Alive",
+            "User-Agent": "okhttp/3.8.1",
+        }
 
-    Args:
+        try:
-        device_id (str): UTE Device id
+            if self.authorization:
+                headers["Authorization"] = f"Bearer {self.authorization}"
+        except AttributeError:
+            pass
 
-    Returns:
+        if method == "GET":
-        dict: UTE account info
+            return requests.get(url, headers=headers)
-    """
 
-    accounts_by_id_url = f"{BASE_URL}/accounts/{device_id}"
+        if method == "POST":
-    return make_request("GET", accounts_by_id_url, authorization=authorization).json()["data"]
+            return requests.post(url, headers=headers, json=data)
 
+        raise Exception("Method not supported")
 
-def get_ute_peak_info(device_id: str, authorization: str) -> dict:
+    def _login(self) -> str:
-    """
+        """
-    Get UTE peak info from device id
+        Login to UTE
 
-    Args:
+        Args:
-        device_id (str): UTE Device id
+            email (str): User email for authentication
+            phone_number (str): User phone number for authentication
 
-    Returns:
+        Returns:
-        dict: UTE peak info
+            str: Authorization token
-    """
+        """
 
-    peak_by_id_url = f"{BASE_URL}/accounts/{device_id}/peak"
+        url = f"{BASE_URL}/token"
-    return make_request("GET", peak_by_id_url, authorization=authorization).json()["data"]
+        data = {
+            "Email": self.email,
+            "PhoneNumber": self.phone_number,
+        }
 
+        return self._make_request("POST", url, data=data).text
 
-def get_ute_network_status(authorization: str) -> dict:
+    def get_devices_list(self) -> List[dict]:
-    """
+        """
-    Get UTE network status from device id
+        Get UTE devices list
 
-    Returns:
+        Returns:
-        dict: UTE network status
+            List[dict]: List of devices
-    """
+        """
 
-    network_status_url = f"{BASE_URL}/info/network/status"
+        accounts_url = f"{BASE_URL}/accounts"
-    return make_request("GET", network_status_url, authorization=authorization).json()["data"]["summary"]
+        return self._make_request("GET", accounts_url).json()["data"]
 
+    def get_account(self) -> dict:
+        """
+        Get UTE account info from device id
 
-def get_ute_renewable_sources(authorization: str) -> str:
+        Returns:
-    """
+            dict: UTE account information
-    Get UTE renewable sources
+        """
 
-    Returns:
+        accounts_by_id_url = f"{BASE_URL}/accounts/{self.device_id}"
-        str: UTE renewable sources percentage
+        return self._make_request("GET", accounts_by_id_url).json()["data"]
-    """
 
-    global_demand_url = f"{BASE_URL}/info/demand/global"
+    def get_peak(self) -> dict:
-    return make_request("GET", global_demand_url).json()["data"]["renewableSources"]
+        """
+        Get UTE peak info from device id
 
+        Returns:
+            dict: UTE peak info
+        """
 
-def get_ute_historic_info(
+        peak_by_id_url = f"{BASE_URL}/accounts/{self.device_id}/peak"
-    device_id: str,
+        return self._make_request("GET", peak_by_id_url).json()["data"]
-    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
 
-    Args:
+    def get_network_status(self) -> List[dict]:
-        device_id (str): UTE Device id
+        """
-        authorization (str): Authorization token
+        Get UTE network status from device id
-        cost_per_kwh (float): Cost per kwh
-        date_start (str): Start date to check
-        date_end (str): End date to check
 
-    Returns:
+        Returns:
-        dict: UTE info
+            dict: UTE network status
-    """
+        """
 
-    if date_start is None:
+        network_status_url = f"{BASE_URL}/info/network/status"
-        yesterday = datetime.now() - timedelta(days=1)
+        return self._make_request("GET", network_status_url).json()["data"]["summary"]
-        date_start = yesterday.strftime("%Y-%m-%d")
 
-    if date_end is None:
+    def get_renewable_sources(self) -> str:
-        yesterday = datetime.now() - timedelta(days=1)
+        """
-        date_end = yesterday.strftime("%Y-%m-%d")
+        Get UTE renewable sources
 
-    historic_url = f"https://rocme.ute.com.uy/api/v2/device/{device_id}/curvefromtodate/D/{date_start}/{date_end}"
+        Returns:
+            str: UTE renewable sources percentage
+        """
 
-    response = make_request("GET", historic_url, authorization=authorization).json()
+        global_demand_url = f"{BASE_URL}/info/demand/global"
+        return self._make_request("GET", global_demand_url).json()["data"]["renewableSources"]
 
-    active_energy = {"total": {"sum_in_kwh": 0}}
+    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
 
-    for item in response["data"]:
+        Args:
-        if item["magnitudeVO"] == "IMPORT_ACTIVE_ENERGY":
+            date_start (str): Start date to check in format YYYY-MM-DD
-            date = datetime.strptime(item["date"], "%Y-%m-%dT%H:%M:%S%z")
+            date_end (str): End date to check in format YYYY-MM-DD
-            day_in_week = date.strftime("%A")
-            value = round(float(item["value"]), 3)
 
-            active_energy[date.strftime("%d/%m/%Y")] = {
+        Returns:
-                "kwh": value,
+            dict: UTE info
-                "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)
+        if date_start 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_start = yesterday.strftime("%Y-%m-%d")
-    )
-    return active_energy
 
+        if date_end is None:
+            yesterday = datetime.now() - timedelta(days=1)
+            date_end = yesterday.strftime("%Y-%m-%d")
 
-def get_current_usage_info(device_id: str, authorization: str) -> dict:
+        historic_url = (
-    """
+            f"https://rocme.ute.com.uy/api/v2/device/{self.device_id}/curvefromtodate/D/{date_start}/{date_end}"
-    Get current usage info from device id
+        )
 
-    Args:
+        response = self._make_request("GET", historic_url).json()
-        device_id (str): UTE Device id
-        authorization (str): Authorization token
 
-    Returns:
+        active_energy = {"total": {"sum_in_kwh": 0}}
-        dict: UTE info
 
-    Raises:
+        for item in response["data"]:
-        Exception: If the reading request fails
+            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)
 
-    reading_request_url = f"{BASE_URL}/device/readingRequest"
+                active_energy[date.strftime("%d/%m/%Y")] = {
-    reading_url = f"{BASE_URL}/device/{device_id}/lastReading/30"
+                    "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
 
-    data = {"AccountServicePointId": 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
 
-    reading_request = make_request("POST", reading_request_url, authorization=authorization, data=data)
+    def _convert_triphasic_powers_to_power_in_watts(self, readings: List[dict]) -> float:
+        """
+        Convert triphasic powers to power in watts
 
-    if reading_request.status_code != 200:
+        Args:
-        raise Exception("Error getting reading request")
+            readings (List[dict]): List of readings
 
-    response = make_request("GET", reading_url, authorization=authorization).json()
+        Returns:
+            float: Power in watts
+        """
 
-    while not response["success"]:
+        for reading in readings:
-        sleep(5)
+            reading_type = reading["tipoLecturaMGMI"]
-        response = make_request("GET", reading_url, authorization=authorization).json()
+            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"])
 
-    readings = response["data"]["readings"]
+        power_1_in_watts = v1 * i1
+        power_2_in_watts = v2 * i2
+        power_3_in_watts = v3 * i3
 
-    for reading in readings:
+        return round(power_1_in_watts + power_2_in_watts + power_3_in_watts, 3)
-        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
+    def get_current_usage_info(self) -> dict:
-    power_2_in_watts = v2 * i2
+        """
-    power_3_in_watts = v3 * i3
+        Get current usage info from device id
 
-    power_in_watts = round(power_1_in_watts + power_2_in_watts + power_3_in_watts, 3)
+        Args:
+            device_id (str): UTE Device id
+            authorization (str): Authorization token
 
-    return_dict = {**response}
+        Returns:
-    return_dict["data"]["power_in_watts"] = power_in_watts
+            dict: UTE info
 
-    return return_dict
+        Raises:
+            Exception: If the reading request fails
+        """
+
+        reading_request_url = f"{BASE_URL}/device/readingRequest"
+        reading_url = f"{BASE_URL}/device/{self.device_id}/lastReading/30"
+
+        data = {"AccountServicePointId": self.device_id}
+
+        reading_request = self._make_request("POST", reading_request_url, data=data)
+
+        if reading_request.status_code != 200:
+            raise Exception("Error getting reading request")
+
+        response = self._make_request("GET", reading_url).json()
+
+        while not response["success"]:
+            sleep(5)
+            response = self._make_request("GET", reading_url).json()
+
+        readings = response["data"]["readings"]
+
+        power_in_watts = self._convert_triphasic_powers_to_power_in_watts(readings)
+
+        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")

src/ute_wrapper/utils.py

@@ -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")