Updated data fetch function, added all data fetch function, refactored

README.md

@@ -3,7 +3,13 @@
 
 ## Confirmed heat pumps that API supports:
 * Thermia Diplomat / Diplomat Duo
-* Thermia iTec (still to be confirmed)
+* Thermia iTec
+
+## Available functions in Thermia class:
+| Function | Description |
+| --- | --- |
+| `fetch_heat_pumps` | Fetches all heat pumps from Thermia Online API and their data |
+| `update_data` | Updates all heat pump data |
 
 ## Available properties within ThermiaHeatPump class:
 | Property | Description |
@@ -28,6 +34,6 @@
 ## Available functions within ThermiaHeatPump class:
 | Function | Description |
 | --- | --- |
-| `refetch_data` | Refetch all data from Thermia for Heat Pump |
+| `update_data` | Refetch all data from Thermia for Heat Pump |
 | `set_temperature` | Set the target temperature for the Heat Pump |
 | `set_operation_mode` | Set the operation mode for the Heat Pump |

ThermiaOnlineAPI/__init__.py

@@ -1,17 +1,26 @@
 from ThermiaOnlineAPI.api.ThermiaAPI import ThermiaAPI
 from ThermiaOnlineAPI.model.HeatPump import ThermiaHeatPump
 
-class Thermia():
+
+class Thermia:
     def __init__(self, username, password):
+        self._username = username
+        self._password = password
+
         self.api_interface = ThermiaAPI(username, password)
         self.connected = self.api_interface.authenticated
-        self.heat_pumps = self.__get_heat_pumps()
 
-    def __get_heat_pumps(self):
+        self.heat_pumps = self.fetch_heat_pumps()
+
+    def fetch_heat_pumps(self) -> list[ThermiaHeatPump]:
         devices = self.api_interface.get_devices()
         heat_pumps = []
 
         for device in devices:
             heat_pumps.append(ThermiaHeatPump(device, self.api_interface))
 
         return heat_pumps
+
+    def update_data(self) -> None:
+        for heat_pump in self.heat_pumps:
+            heat_pump.update_data()

ThermiaOnlineAPI/api/ThermiaAPI.py

@@ -2,15 +2,23 @@
 from datetime import datetime
 import requests
 
+
+from ..exceptions.AuthenticationException import AuthenticationException
+from ..exceptions.NetworkException import NetworkException
 from ..model.HeatPump import ThermiaHeatPump
 
 LOGGER = logging.getLogger(__name__)
 
-DEFAULT_REQUEST_HEADERS = {"Authorization": "Bearer %s", "Content-Type": "application/json"}
+DEFAULT_REQUEST_HEADERS = {
+    "Authorization": "Bearer %s",
+    "Content-Type": "application/json",
+}
 
 THERMIA_API_CONFIG_URL = "https://online.thermia.se/api/configuration"
+THERMIA_INSTALLATION_PATH = "/api/v1/Registers/Installations/"
+
 
-class ThermiaAPI():
+class ThermiaAPI:
     def __init__(self, email, password):
         self.__email = email
         self.__password = password
@@ -23,7 +31,7 @@ def __init__(self, email, password):
     def get_devices(self):
         self.__check_token_validity()
 
-        url = self.configuration['apiBaseUrl'] + "/api/v1/InstallationsInfo/own"
+        url = self.configuration["apiBaseUrl"] + "/api/v1/InstallationsInfo/own"
         request = requests.get(url, headers=DEFAULT_REQUEST_HEADERS)
         status = request.status_code
         LOGGER.info("Fetching devices. " + str(status))
@@ -37,7 +45,11 @@ def get_devices(self):
     def get_device_info(self, device):
         self.__check_token_validity()
 
-        url = self.configuration['apiBaseUrl'] + "/api/v1/installations/" + str(device['id'])
+        url = (
+            self.configuration["apiBaseUrl"]
+            + "/api/v1/installations/"
+            + str(device["id"])
+        )
         request = requests.get(url, headers=DEFAULT_REQUEST_HEADERS)
         status = request.status_code
 
@@ -50,7 +62,12 @@ def get_device_info(self, device):
     def get_device_status(self, device):
         self.__check_token_validity()
 
-        url = self.configuration['apiBaseUrl'] + "/api/v1/installationstatus/" + str(device['id']) + "/status"
+        url = (
+            self.configuration["apiBaseUrl"]
+            + "/api/v1/installationstatus/"
+            + str(device["id"])
+            + "/status"
+        )
         request = requests.get(url, headers=DEFAULT_REQUEST_HEADERS)
         status = request.status_code
 
@@ -63,7 +80,12 @@ def get_device_status(self, device):
     def get_temperature_status(self, device: ThermiaHeatPump):
         self.__check_token_validity()
 
-        url = self.configuration['apiBaseUrl'] + "/api/v1/Registers/Installations/" + str(device.id) + "/Groups/REG_GROUP_TEMPERATURES"
+        url = (
+            self.configuration["apiBaseUrl"]
+            + THERMIA_INSTALLATION_PATH
+            + str(device.id)
+            + "/Groups/REG_GROUP_TEMPERATURES"
+        )
         request = requests.get(url, headers=DEFAULT_REQUEST_HEADERS)
         status = request.status_code
 
@@ -73,10 +95,16 @@ def get_temperature_status(self, device: ThermiaHeatPump):
 
         device_temperature_register_index = device.get_register_indexes()["temperature"]
         if device_temperature_register_index is None:
-            LOGGER.error("Error in getting device's temperature status. No temperature register index.")
+            LOGGER.error(
+                "Error in getting device's temperature status. No temperature register index."
+            )
             return None
 
-        data = [d for d in request.json() if d['registerIndex'] == device_temperature_register_index]
+        data = [
+            d
+            for d in request.json()
+            if d["registerIndex"] == device_temperature_register_index
+        ]
 
         if len(data) == 0:
             # Temperature status not supported
@@ -85,104 +113,147 @@ def get_temperature_status(self, device: ThermiaHeatPump):
         data = data[0]
 
         return {
-            "minValue": data['minValue'],
-            "maxValue": data['maxValue'],
-            "step": data['step'],
+            "minValue": data["minValue"],
+            "maxValue": data["maxValue"],
+            "step": data["step"],
         }
 
     def get_operation_mode(self, device: ThermiaHeatPump):
         self.__check_token_validity()
 
-        url = self.configuration['apiBaseUrl'] + "/api/v1/Registers/Installations/" + str(device.id) + "/Groups/REG_GROUP_OPERATIONAL_OPERATION"
+        url = (
+            self.configuration["apiBaseUrl"]
+            + THERMIA_INSTALLATION_PATH
+            + str(device.id)
+            + "/Groups/REG_GROUP_OPERATIONAL_OPERATION"
+        )
         request = requests.get(url, headers=DEFAULT_REQUEST_HEADERS)
         status = request.status_code
 
         if status != 200:
             LOGGER.error("Error in getting device's operation mode. " + str(status))
             return None
 
-        data = [d for d in request.json() if d['registerName'] == "REG_OPERATIONMODE"]
+        data = [d for d in request.json() if d["registerName"] == "REG_OPERATIONMODE"]
 
         if len(data) == 0:
             # Operation mode not supported
             return None
 
         data = data[0]
 
-        device.set_register_index_operation_mode(data['registerIndex'])
+        device.set_register_index_operation_mode(data["registerIndex"])
 
         current_operation_mode = int(data.get("registerValue"))
         operation_modes_data = data.get("valueNames")
 
         if operation_modes_data is not None:
-            operation_modes = list(map(lambda values: values.get("name").split("REG_VALUE_OPERATION_MODE_")[1], operation_modes_data))
+            operation_modes = list(
+                map(
+                    lambda values: values.get("name").split(
+                        "REG_VALUE_OPERATION_MODE_"
+                    )[1],
+                    operation_modes_data,
+                )
+            )
             return {
                 "current": operation_modes[current_operation_mode],
-                "available": operation_modes
+                "available": operation_modes,
             }
 
         return None
 
     def set_temperature(self, device: ThermiaHeatPump, temperature):
         device_temperature_register_index = device.get_register_indexes()["temperature"]
         if device_temperature_register_index is None:
-            LOGGER.error("Error setting device's temperature. No temperature register index.")
+            LOGGER.error(
+                "Error setting device's temperature. No temperature register index."
+            )
             return
 
-        self.__set_register_value(device, device_temperature_register_index, temperature)
+        self.__set_register_value(
+            device, device_temperature_register_index, temperature
+        )
 
     def set_operation_mode(self, device: ThermiaHeatPump, mode):
         operation_mode_int = device.available_operation_modes.index(mode)
 
-        device_operation_mode_register_index = device.get_register_indexes()["operation_mode"]
+        device_operation_mode_register_index = device.get_register_indexes()[
+            "operation_mode"
+        ]
         if device_operation_mode_register_index is None:
-            LOGGER.error("Error setting device's operation mode. No operation mode register index.")
+            LOGGER.error(
+                "Error setting device's operation mode. No operation mode register index."
+            )
             return
-
-        self.__set_register_value(device, device_operation_mode_register_index, operation_mode_int)
 
-    def __set_register_value(self, device: ThermiaHeatPump, register_index: int, register_value: int):
+        self.__set_register_value(
+            device, device_operation_mode_register_index, operation_mode_int
+        )
+
+    def __set_register_value(
+        self, device: ThermiaHeatPump, register_index: int, register_value: int
+    ):
         self.__check_token_validity()
 
-        url = self.configuration['apiBaseUrl'] + "/api/v1/Registers/Installations/" + str(device.id) + "/Registers"
+        url = (
+            self.configuration["apiBaseUrl"]
+            + THERMIA_INSTALLATION_PATH
+            + str(device.id)
+            + "/Registers"
+        )
         body = {
             "registerIndex": register_index,
             "registerValue": register_value,
-            "clientUuid": "api-client-uuid"
+            "clientUuid": "api-client-uuid",
         }
 
-        request =  requests.post(url, headers=DEFAULT_REQUEST_HEADERS, json=body)
+        request = requests.post(url, headers=DEFAULT_REQUEST_HEADERS, json=body)
 
         status = request.status_code
         if status != 200:
-            LOGGER.error("Error setting register " + str(register_index) + " value. " + str(status))
+            LOGGER.error(
+                "Error setting register "
+                + str(register_index)
+                + " value. "
+                + str(status)
+            )
 
     def __fetch_configuration(self):
         request = requests.get(THERMIA_API_CONFIG_URL)
         status = request.status_code
 
         if status != 200:
             LOGGER.error("Error fetching API configuration. " + str(status))
-            raise Exception("Error fetching API configuration.", status)
+            raise NetworkException("Error fetching API configuration.", status)
 
         return request.json()
 
     def __authenticate(self):
-        auth_url = self.configuration['authApiBaseUrl'] + "/api/v1/Jwt/login"
-        json = {"userName": self.__email, "password": self.__password, "rememberMe": True}
+        auth_url = self.configuration["authApiBaseUrl"] + "/api/v1/Jwt/login"
+        json = {
+            "userName": self.__email,
+            "password": self.__password,
+            "rememberMe": True,
+        }
 
         request_auth = requests.post(auth_url, json=json)
         status = request_auth.status_code
 
         if status != 200:
-            LOGGER.error("Authentication request failed, please check credentials. " + str(status))
-            raise Exception("Authentication request failed, please check credentials.", status)
+            LOGGER.error(
+                "Authentication request failed, please check credentials. "
+                + str(status)
+            )
+            raise AuthenticationException(
+                "Authentication request failed, please check credentials.", status
+            )
 
         auth_data = request_auth.json()
         LOGGER.debug(str(auth_data))
 
         token_valid_to = auth_data.get("tokenValidToUtc").split(".")[0]
-        datetime_object = datetime.strptime(token_valid_to, '%Y-%m-%dT%H:%M:%S')
+        datetime_object = datetime.strptime(token_valid_to, "%Y-%m-%dT%H:%M:%S")
         token_valid_to = datetime_object.timestamp()
 
         self.__token = auth_data.get("token")
@@ -196,6 +267,9 @@ def __authenticate(self):
         return True
 
     def __check_token_validity(self):
-        if self.__token_valid_to is None or self.__token_valid_to < datetime.now().timestamp():
+        if (
+            self.__token_valid_to is None
+            or self.__token_valid_to < datetime.now().timestamp()
+        ):
             LOGGER.info("Token expired, reauthenticating.")
-            self.authenticated = self.__authenticate()
+            self.authenticated = self.__authenticate()

ThermiaOnlineAPI/exceptions/AuthenticationException.py

@@ -0,0 +1,10 @@
+class AuthenticationException(Exception):
+    """
+    Exception raised when the authentication fails.
+    """
+
+    def __init__(self, message, status=None):
+
+        super().__init__(message)
+
+        self.status = status

ThermiaOnlineAPI/exceptions/NetworkException.py

@@ -0,0 +1,10 @@
+class NetworkException(Exception):
+    """
+    Exception raised when the network fails.
+    """
+
+    def __init__(self, message, status=None):
+
+        super().__init__(message)
+
+        self.status = status

ThermiaOnlineAPI/exceptions/__init__.py

@@ -0,0 +1 @@
+"""Thermia API Exceptions"""