Add gateway support (#1)

* Add support for the gateway

* Battery sensor added

* Optimze imports

* Less API calls, type hints

* README update

README.md

@@ -4,13 +4,22 @@ A custom integration which will add your thermostats to Home Assistant.
 ## Pre-install
 1. Create a user for Home Assistant in the control box. Do not use a user that's in use by any other application. For example, the mobile app or web app.
 2. Make sure Home Assistant is able to reach the control box via the local network.
+2. Make sure Home Assistant is able to reach the gateway via the local network.
+3. You need to have the password for the gateway.
 
 ## Install
-1. Download the zip file from Github.
+1. Download the latest version release from Github.
 2. Upload the files into the directory `/config/custom_components`.
 3. Rename the directory to `alpha`. The files should exist in the directory `/config/custom_components/alpha`.
 4. Restart Home Assistant
 5. Now you should be able to add the integration via UI.
+6. Required information:
+   1. Gateway IP
+   2. Gateway password
+   3. Control box IP
+   4. Control box username
+   5. Control box password
 
 ## Disclaimer
-I cannot say that i'm a python developer. So, if you see code that's bad practice. Please, feel free to contribute to this integration via a pull request.
+- I cannot say that i'm a python developer. So, if you see code that's bad practice. Please, feel free to contribute to this integration via a pull request. 
+- I do not own an Alpha Innotect heat pump. If you have an issue, please provide sample data and information about your installation because I can't easily test changes.

__init__.py

@@ -3,21 +3,26 @@
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 
-from .api import ControllerApi
 from .const import DOMAIN, PLATFORMS
+from .controller_api import ControllerAPI
+from .gateway_api import GatewayAPI
 
 _LOGGER = logging.getLogger(__name__)
 
 
-async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> True:
     """Set up Alpha Home from config entry."""
     _LOGGER.debug("Setting up Alpha Home component")
 
-    controller_api = ControllerApi(entry.data['controller_ip'], entry.data['username'], entry.data['password'])
-
+    controller_api = ControllerAPI(entry.data['controller_ip'], entry.data['controller_username'], entry.data['controller_password'])
     controller_api = await hass.async_add_executor_job(controller_api.login)
+    gateway_api = GatewayAPI(entry.data['gateway_ip'], entry.data['gateway_password'])
+    gateway_api = await hass.async_add_executor_job(gateway_api.login)
 
-    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = controller_api
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
+        "controller_api": controller_api,
+        "gateway_api": gateway_api,
+    }
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 

api.py

@@ -1,32 +1,29 @@
 import base64
 import logging
-import urllib
-from urllib.parse import unquote
 
-import requests
 from Crypto.Cipher import AES
 from Crypto.Hash import SHA256
 from backports.pbkdf2 import pbkdf2_hmac
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class ControllerApi:
+class BaseAPI:
 
-    def __init__(self, hostname, username, password) -> None:
-        self.username = username
-        self.password = password
-        self.api_host = hostname
+    def __init__(self, hostname: str, username: str, password: str) -> None:
+        self.api_host: str = hostname
+        self.username: str = username
+        self.password: str = password
 
-        self.user_id = None
-        self.device_token_encrypted = None
-        self.device_token_decrypted = None
-        self.request_count = 0
-        self.last_request_signature = None
-        self.udid = "homeassistant"
+        self.user_id: int | None = None
+        self.device_token_encrypted: str | None = None
+        self.device_token_decrypted: str | None = None
+        self.request_count: int = 0
+        self.last_request_signature: str | None = None
+        self.udid: str = "homeassistant"
 
     @staticmethod
-    def string_to_charcodes(data: str):
+    def string_to_charcodes(data: str) -> str:
         a = ""
         if len(data) > 0:
             for i in range(len(data)):
@@ -45,156 +42,20 @@ def encode_signature(self, value: str, salt: str) -> str:
 
         return original
 
-    def login(self):
-        response = requests.post("http://" + self.api_host + "/api/user/token/challenge", data={
-            "udid": self.udid
-        })
-
-        device_token = response.json()['devicetoken']
-
-        response = requests.post("http://" + self.api_host + "/api/user/token/response", data={
-            "login": self.username,
-            "token": device_token,
-            "udid": self.udid,
-            "hashed": base64.b64encode(self.encode_signature(self.password, device_token)).decode()
-        })
-
-        self.device_token_encrypted = response.json()['devicetoken_encrypted']
-        self.user_id = response.json()['userid']
-
-        self.device_token_decrypted = self.decrypt2(response.json()['devicetoken_encrypted'], self.password)
-
-        response = self.call("admin/login/check")
-
-        if not response['success']:
-            raise Exception("Unable to login")
-
-        return self
-
     @staticmethod
-    def _prepare_request_body_for_hash(urlencoded_string):
-        urlencoded_string = urlencoded_string.replace('%2C', ',').replace('%5B', '[').replace('%5D', ']')
+    def _prepare_request_body_for_hash(urlencoded_string: str) -> str:
         """Replace dots for comma in case of temperature being passed"""
+        urlencoded_string = urlencoded_string.replace('%2C', ',').replace('%5B', '[').replace('%5D', ']')
         # if(urlencodedString.find("temperature") != -1):
         #    urlencodedString = urlencodedString.replace('.', ',')
         return urlencoded_string
 
     @staticmethod
-    def decrypt2(encrypted_data, key):
+    def decrypt2(encrypted_data: str, key: str):
         static_iv = 'D3GC5NQEFH13is04KD2tOg=='
         crypt_key = SHA256.new()
         crypt_key.update(bytes(key, 'utf-8'))
         crypt_key = crypt_key.digest()
         cipher = AES.new(crypt_key, AES.MODE_CBC, base64.b64decode(static_iv))
 
         return cipher.decrypt(base64.b64decode(encrypted_data)).decode('ascii').strip('\x10')
-
-    def call(self, endpoint: str, data: dict = {}):
-        _LOGGER.debug("Requesting: %s", endpoint)
-        json_response = None
-
-        try:
-            data['userid'] = self.user_id
-            data['udid'] = self.udid
-            data['reqcount'] = self.request_count
-
-            post_data_sorted = sorted(data.items(), key=lambda val: val[0])
-
-            urlencoded_body = urllib.parse.urlencode(post_data_sorted, encoding='utf-8')
-            urlencoded_body_prepared_for_hash = self._prepare_request_body_for_hash(urlencoded_body)
-            urlencoded_body_prepared_for_hash = urlencoded_body_prepared_for_hash.replace('&', '|')
-            urlencoded_body_prepared_for_hash = urlencoded_body_prepared_for_hash + "|"
-
-            request_signature = base64.b64encode(
-                self.encode_signature(urlencoded_body_prepared_for_hash, self.device_token_decrypted)).decode()
-
-            self.last_request_signature = request_signature
-
-            urlencoded_body = urlencoded_body + "&" + urllib.parse.urlencode({"request_signature": request_signature},
-                                                                             encoding='utf-8')
-
-            _LOGGER.debug("Encoded body: %s", urlencoded_body)
-
-            response = requests.post("http://{hostname}/{endpoint}".format(hostname=self.api_host, endpoint=endpoint),
-                                     data=urlencoded_body,
-                                     headers={'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'}
-                                     )
-
-            self.request_count = self.request_count + 1
-            json_response = response.json()
-        except Exception as exception:
-            _LOGGER.exception("Unable to fetch data from API: %s", exception)
-
-        _LOGGER.debug("Response: %s", json_response)
-
-        if not json_response['success']:
-            raise Exception('Failed to get data')
-        else:
-            _LOGGER.debug('Successfully fetched data from API')
-
-        return json_response
-
-    def room_list(self):
-        return self.call("api/room/list")
-
-    def room_details(self, identifier):
-        room_list = self.room_list()
-        for group in room_list['groups']:
-            for room in group['rooms']:
-                if room['id'] == identifier:
-                    return room
-
-        return None
-
-    def system_information(self):
-        return self.call('admin/systeminformation/get')
-
-    def set_temperature(self, room_identifier, temperature: float):
-        return self.call('api/room/settemperature', {
-            "roomid": room_identifier,
-            "temperature": temperature
-        })
-
-    def thermostats(self):
-        thermostats: list[Thermostat] = []
-
-        try:
-            response = self.room_list()
-            for group in response['groups']:
-                for room in group['rooms']:
-                    thermostat = Thermostat(
-                        identifier=room['id'],
-                        module="Test",
-                        name=room['name'],
-                        current_temperature=room.get('actualTemperature'),
-                        desired_temperature=room.get('desiredTemperature'),
-                        minimum_temperature=room.get('minTemperature'),
-                        maximum_temperature=room.get('minTemperature'),
-                        cooling=room.get('cooling'),
-                        cooling_enabled=room.get('coolingEnabled')
-                    )
-
-                    thermostats.append(thermostat)
-        except Exception as exception:
-            _LOGGER.exception("There is an exception: %s", exception)
-
-        return thermostats
-
-
-class Thermostat:
-    def __init__(self, identifier: str, name: str, module: str = None, current_temperature: float = None,
-                 minimum_temperature: float = None, maximum_temperature: float = None,
-                 desired_temperature: float = None, battery_percentage: int = None,
-                 cooling: bool = None,
-                 cooling_enabled: bool = False
-                 ):
-        self.identifier = identifier
-        self.module = module
-        self.name = name
-        self.current_temperature = current_temperature
-        self.minimum_temperature = minimum_temperature
-        self.maximum_temperature = maximum_temperature
-        self.battery_percentage = battery_percentage
-        self.desired_temperature = desired_temperature
-        self.cooling = cooling
-        self.cooling_enabled = cooling_enabled

base_coordinator.py

@@ -0,0 +1,64 @@
+"""Platform for sensor integration."""
+from __future__ import annotations
+
+import logging
+
+from homeassistant.core import HomeAssistant
+
+from . import GatewayAPI
+from .const import MODULE_TYPE_SENSOR
+from .controller_api import ControllerAPI
+from .structs.Thermostat import Thermostat
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class BaseCoordinator:
+
+    @staticmethod
+    async def get_thermostats(hass: HomeAssistant, gateway_api: GatewayAPI, controller_api: ControllerAPI) -> list[Thermostat]:
+        try:
+            rooms: dict = await hass.async_add_executor_job(gateway_api.all_modules)
+
+            thermostats: list[Thermostat] = []
+
+            db_modules: dict = await hass.async_add_executor_job(gateway_api.db_modules)
+            room_list: dict = await hass.async_add_executor_job(controller_api.room_list)
+
+            try:
+                for room_id in rooms:
+                    room_module = rooms[room_id]
+                    room = await hass.async_add_executor_job(controller_api.room_details, room_id, room_list)
+
+                    current_temperature = None
+                    battery_percentage = None
+
+                    for module_id in room_module['modules']:
+                        if module_id not in db_modules['modules']:
+                            continue
+
+                        module_details = db_modules['modules'][module_id]
+
+                        if module_details["type"] == MODULE_TYPE_SENSOR:
+                            current_temperature = module_details["currentTemperature"]
+                            battery_percentage = module_details["battery"]
+
+                    thermostat = Thermostat(
+                        identifier=room_id,
+                        name=room['name'],
+                        current_temperature=current_temperature,
+                        desired_temperature=room.get('desiredTemperature'),
+                        minimum_temperature=room.get('minTemperature'),
+                        maximum_temperature=room.get('maxTemperature'),
+                        cooling=room.get('cooling'),
+                        cooling_enabled=room.get('coolingEnabled'),
+                        battery_percentage=battery_percentage
+                    )
+
+                    thermostats.append(thermostat)
+            except Exception as exception:
+                _LOGGER.exception("There is an exception: %s", exception)
+
+            return thermostats
+        except Exception as exception:
+            raise exception