Initial commit

.gitignore

@@ -1,129 +1,5 @@
-# Byte-compiled / optimized / DLL files
+# Python
 __pycache__/
-*.py[cod]
-*$py.class
 
-# C extensions
-*.so
-
-# Distribution / packaging
-.Python
-build/
-develop-eggs/
-dist/
-downloads/
-eggs/
-.eggs/
-lib/
-lib64/
-parts/
-sdist/
-var/
-wheels/
-pip-wheel-metadata/
-share/python-wheels/
-*.egg-info/
-.installed.cfg
-*.egg
-MANIFEST
-
-# PyInstaller
-#  Usually these files are written by a python script from a template
-#  before PyInstaller builds the exe, so as to inject date/other infos into it.
-*.manifest
-*.spec
-
-# Installer logs
-pip-log.txt
-pip-delete-this-directory.txt
-
-# Unit test / coverage reports
-htmlcov/
-.tox/
-.nox/
-.coverage
-.coverage.*
-.cache
-nosetests.xml
-coverage.xml
-*.cover
-*.py,cover
-.hypothesis/
-.pytest_cache/
-
-# Translations
-*.mo
-*.pot
-
-# Django stuff:
-*.log
-local_settings.py
-db.sqlite3
-db.sqlite3-journal
-
-# Flask stuff:
-instance/
-.webassets-cache
-
-# Scrapy stuff:
-.scrapy
-
-# Sphinx documentation
-docs/_build/
-
-# PyBuilder
-target/
-
-# Jupyter Notebook
-.ipynb_checkpoints
-
-# IPython
-profile_default/
-ipython_config.py
-
-# pyenv
-.python-version
-
-# pipenv
-#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
-#   However, in case of collaboration, if having platform-specific dependencies or dependencies
-#   having no cross-platform support, pipenv may install dependencies that don't work, or not
-#   install all needed dependencies.
-#Pipfile.lock
-
-# PEP 582; used by e.g. github.com/David-OConnor/pyflow
-__pypackages__/
-
-# Celery stuff
-celerybeat-schedule
-celerybeat.pid
-
-# SageMath parsed files
-*.sage.py
-
-# Environments
-.env
-.venv
-env/
-venv/
-ENV/
-env.bak/
-venv.bak/
-
-# Spyder project settings
-.spyderproject
-.spyproject
-
-# Rope project settings
-.ropeproject
-
-# mkdocs documentation
-/site
-
-# mypy
-.mypy_cache/
-.dmypy.json
-dmypy.json
-
-# Pyre type checker
-.pyre/
+# IDE
+.vscode

ThermiaOnlineAPI/__init__.py

@@ -0,0 +1,17 @@
+from .api.ThermiaAPI import ThermiaAPI
+from .model.WaterHeater import ThermiaWaterHeater
+
+class Thermia():
+    def __init__(self, username, password):
+        self.api_interface = ThermiaAPI(username, password)
+        self.connected = self.api_interface.authenticated
+        self.water_heaters = self.__get_water_heaters()
+
+    def __get_water_heaters(self):
+        devices = self.api_interface.get_devices()
+        water_heaters = []
+
+        for device in devices:
+            water_heaters.append(ThermiaWaterHeater(device, self.api_interface))
+
+        return water_heaters

ThermiaOnlineAPI/api/ThermiaAPI.py

@@ -0,0 +1,157 @@
+import logging
+from datetime import datetime
+import requests
+
+from ..model.WaterHeater import ThermiaWaterHeater
+
+LOGGER = logging.getLogger(__name__)
+
+DEFAULT_REQUEST_HEADERS = {"Authorization": "Bearer %s", "Content-Type": "application/json"}
+
+THERMIA_API_CONFIG_URL = "https://online.thermia.se/api/configuration"
+
+SET_REGISTER_VALUES = {
+    "set_temperature": 50,
+    "set_operation_mode": 51,
+}
+
+class ThermiaAPI():
+    def __init__(self, email, password):
+        self.__email = email
+        self.__password = password
+        self.__token = None
+        self.__token_valid_to = None
+
+        self.configuration = self.__fetch_configuration()
+        self.authenticated = self.__authenticate()
+
+    def get_devices(self):
+        self.__check_token_validity()
+
+        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))
+
+        if status != 200:
+            LOGGER.error("Error fetching devices. " + str(status))
+            return []
+
+        return request.json()
+
+    def get_device_info(self, device):
+        self.__check_token_validity()
+
+        url = self.configuration['apiBaseUrl'] + "/api/v1/installations/" + str(device['id'])
+        request = requests.get(url, headers=DEFAULT_REQUEST_HEADERS)
+        status = request.status_code
+
+        if status != 200:
+            LOGGER.error("Error fetching device info. " + str(status))
+            return None
+
+        return request.json()
+
+    def get_device_status(self, device):
+        self.__check_token_validity()
+
+        url = self.configuration['apiBaseUrl'] + "/api/v1/installationstatus/" + str(device['id']) + "/status"
+        request = requests.get(url, headers=DEFAULT_REQUEST_HEADERS)
+        status = request.status_code
+
+        if status != 200:
+            LOGGER.error("Error fetching device status. " + str(status))
+            return None
+
+        return request.json()
+
+    def get_operation_mode(self, device):
+        self.__check_token_validity()
+
+        url = self.configuration['apiBaseUrl'] + "/api/v1/Registers/Installations/" + 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 = request.json()[0]
+
+        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))
+            return {
+                "current": operation_modes[current_operation_mode],
+                "available": operation_modes
+            }
+
+        return None
+
+    def set_temperature(self, device: ThermiaWaterHeater, temperature):
+        self.__set_register_value(device, SET_REGISTER_VALUES["set_temperature"], temperature)
+
+    def set_operation_mode(self, device: ThermiaWaterHeater, mode):
+        operation_mode_int = device.available_operation_modes.index(mode)
+        self.__set_register_value(device, SET_REGISTER_VALUES["set_operation_mode"], operation_mode_int)
+
+    def __set_register_value(self, device: ThermiaWaterHeater, register_index: SET_REGISTER_VALUES, register_value: int):
+        self.__check_token_validity()
+
+        url = self.configuration['apiBaseUrl'] + "/api/v1/Registers/Installations/" + str(device.id) + "/Registers"
+        body = {
+            "registerIndex": register_index,
+            "registerValue": register_value,
+            "clientUuid": "api-client-uuid"
+        }
+
+        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))
+
+    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)
+
+        return request.json()
+
+    def __authenticate(self):
+        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)
+
+        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')
+        token_valid_to = datetime_object.timestamp()
+
+        self.__token = auth_data.get("token")
+        self.__token_valid_to = token_valid_to
+
+        auth = DEFAULT_REQUEST_HEADERS.get("Authorization")
+        auth = auth % self.__token
+        DEFAULT_REQUEST_HEADERS["Authorization"] = auth
+
+        LOGGER.info("Authentication was successful, token set.")
+        return True
+
+    def __check_token_validity(self):
+        if self.__token_valid_to is None or self.__token_valid_to < datetime.now().timestamp():
+            LOGGER.info("Token expired, reauthenticating.")
+            self.authenticated = self.__authenticate()

ThermiaOnlineAPI/model/WaterHeater.py

@@ -0,0 +1,86 @@
+import json
+import logging
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ..api.ThermiaAPI import ThermiaAPI
+
+LOGGER = logging.getLogger(__name__)
+
+class ThermiaWaterHeater():
+    def __init__(self, device_data: json, api_interface: "ThermiaAPI"):
+        self.__device_data = device_data
+        self.__api_interface = api_interface
+        self.__info = None
+        self.__status = None
+        self.__operation_mode_state = None
+
+        self.refetch_data()
+
+    def refetch_data(self):
+        self.__info = self.__api_interface.get_device_info(self.__device_data)
+        self.__status = self.__api_interface.get_device_status(self.__device_data)
+        self.__operation_mode_state = self.__api_interface.get_operation_mode(self.__device_data)
+
+    def set_temperature(self, temperature: int):
+        LOGGER.info("Setting temperature to " + str(temperature))
+        self.__api_interface.set_temperature(self, temperature)
+        self.refetch_data()
+
+    def set_operation_mode(self, mode: str):
+        LOGGER.info("Setting operation mode to " + str(mode))
+        self.__api_interface.set_operation_mode(self, mode)
+        self.refetch_data()
+
+    @property
+    def name(self):
+        return self.__info.get("name")
+
+    @property
+    def id(self):
+        return self.__info.get("id")
+
+    @property
+    def is_online(self):
+        return self.__info.get("isOnline")
+
+    @property
+    def last_online(self):
+        return self.__info.get("lastOnline")
+
+    @property
+    def has_indoor_temp_sensor(self):
+        return self.__status.get("hasIndoorTempSensor")
+
+    @property
+    def indoor_temperature(self):
+        return self.__status.get("indoorTemperature")
+
+    @property
+    def is_outdoor_temp_sensor_functioning(self):
+        return self.__status.get("isOutdoorTempSensorFunctioning")
+
+    @property
+    def outdoor_temperature(self):
+        return self.__status.get("outdoorTemperature")
+
+    @property
+    def is_hot_water_active(self):
+        return self.__status.get("isHotwaterActive")
+
+    @property
+    def hot_water_temperature(self):
+        return self.__status.get("hotWaterTemperature")
+
+    @property
+    def heat_temperature(self):
+        return self.__status.get("heatingEffect")
+
+    @property
+    def operation_mode(self):
+        return self.__operation_mode_state["current"]
+
+    @property
+    def available_operation_modes(self):
+        return self.__operation_mode_state["available"]