Permit JSON Live Data Server

As per Leo's request on issue 3 #3 amended config_flow.py, sensor.py and resolapi.py

custom_components/resol/config_flow.py

@@ -1,4 +1,5 @@
 """config_flow.py: Config flow for Resol integration."""
+
 from __future__ import annotations
 
 import logging
@@ -15,23 +16,15 @@
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.exceptions import IntegrationError
 
-from .const import (
-    _LOGGER,
-    DOMAIN,
-    ISSUE_URL_ERROR_MESSAGE
-)
+from .const import _LOGGER, DOMAIN, ISSUE_URL_ERROR_MESSAGE
 
-from .resolapi import (
-    ResolAPI,
-    AuthenticationFailed
-)
+from .resolapi import ResolAPI, AuthenticationFailed
 
 
 # This is the first step's schema when setting up the integration, or its devices
 # The second schema is defined inside the ConfigFlow class as it has dynamice default values set via API call
 STEP_USER_DATA_SCHEMA = vol.Schema(
     {
-
         vol.Required("host", default=""): str,
         vol.Required("port", default="80"): str,
         vol.Required("username", default="admin"): str,
@@ -40,8 +33,9 @@
 )
 
 
-
-async def validate_input_for_device(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
+async def validate_input_for_device(
+    hass: HomeAssistant, data: dict[str, Any]
+) -> dict[str, Any]:
     """Validate the user input allows us to connect."""
 
     resol_api = ResolAPI(data["host"], data["port"], data["username"], data["password"])
@@ -51,7 +45,14 @@ async def validate_input_for_device(hass: HomeAssistant, data: dict[str, Any]) -
         device = await hass.async_add_executor_job(resol_api.detect_device)
 
         # Additionally, check for authentication by calling fetch_data_km2
-        auth_check = await hass.async_add_executor_job(resol_api.fetch_data_km2)
+        # As requested here: https://github.com/evercape/hass-resol-KM2/issues/3
+        if device["product"] == "KM2":
+            auth_check = await hass.async_add_executor_job(resol_api.fetch_data_km2)
+        elif device["product"] == "KM1":
+            auth_check = await hass.async_add_executor_job(resol_api.fetch_data_km1)
+        elif device["product"] == "DL2" or device["product"] == "DL3":
+            auth_check = await hass.async_add_executor_job(resol_api.fetch_data_dlx)
+
         if not auth_check:
             # If authentication check returns False, raise an authentication failure exception
             raise AuthenticationFailed("Invalid authentication")
@@ -61,16 +62,17 @@ async def validate_input_for_device(hass: HomeAssistant, data: dict[str, Any]) -
 
     # Exception if device cannot be found
     except IntegrationError as e:
-        _LOGGER.error(f"Failed to connect to Resol device: {e}"+ISSUE_URL_ERROR_MESSAGE)
+        _LOGGER.error(
+            f"Failed to connect to Resol device: {e}" + ISSUE_URL_ERROR_MESSAGE
+        )
         raise CannotConnect from e
 
     # Exception if authentication fails
     except AuthenticationFailed as e:
-        _LOGGER.error(f"Authentication failed: {e}"+ISSUE_URL_ERROR_MESSAGE)
+        _LOGGER.error(f"Authentication failed: {e}" + ISSUE_URL_ERROR_MESSAGE)
         raise InvalidAuth from e
 
 
-
 class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
     """Handle a config flow for Resol."""
 
@@ -108,7 +110,9 @@ async def async_step_user(
 
                 # Checks that the device is actually unique, otherwise abort
                 await self.async_set_unique_id(unique_id)
-                self._abort_if_unique_id_configured(updates={"host": user_input["host"]})
+                self._abort_if_unique_id_configured(
+                    updates={"host": user_input["host"]}
+                )
 
                 # Before creating the entry in the config_entry registry, go to step 2 for the options
                 # However, make sure the steps from the user input are passed on to the next step
@@ -125,7 +129,8 @@ async def async_step_user(
 
     # This is step 2 for the options such as custom name, group and disable sensors
     async def async_step_device_options(
-        self, user_input: dict[str, Any] | None = None,
+        self,
+        user_input: dict[str, Any] | None = None,
     ) -> FlowResult:
         """Handle the device options step."""
 
@@ -134,32 +139,44 @@ async def async_step_device_options(
         if user_input is not None:
             try:
                 # Sanitize the user provided custom device name, which is used for entry and device registry name
-                user_input["custom_device_name"] = sanitize_device_name(user_input["custom_device_name"], self.device_info["name"])
+                user_input["custom_device_name"] = sanitize_device_name(
+                    user_input["custom_device_name"], self.device_info["name"]
+                )
 
                 # Since we have already set the unique ID and updated host if necessary create the entry with the additional options.
                 # The title of the integration is the custom friendly device name given by the user in step 2
                 title = user_input["custom_device_name"]
                 return self.async_create_entry(
                     title=title,
                     data={
-                        "user_input": self.user_input_from_step_user,   # from previous step
-                        "device_info": self.device_info,                # from device detection
-                        "options": user_input                           # new options from this step
+                        "user_input": self.user_input_from_step_user,  # from previous step
+                        "device_info": self.device_info,  # from device detection
+                        "options": user_input,  # new options from this step
                     },
                 )
             except Exception as e:
-                _LOGGER.error(f"Failed to handle device options: {e}"+ISSUE_URL_ERROR_MESSAGE)
+                _LOGGER.error(
+                    f"Failed to handle device options: {e}" + ISSUE_URL_ERROR_MESSAGE
+                )
                 errors["base"] = "option_error"
 
         # Prepare the second form's schema as it has dynamic values based on the API call
         # Use the name from the detected device as default device name
-        default_device_name = self.device_info["name"] if self.device_info and "name" in self.device_info else "New Device"
-        step_device_options_schema = vol.Schema({
-            vol.Required("custom_device_name", default=default_device_name): str,
-            vol.Required("polling_time", default=60): vol.All(vol.Coerce(int), vol.Clamp(min=60)),
-            vol.Required("group_sensors", default=True): bool,
-            vol.Required("disable_sensors", default=True): bool,
-        })
+        default_device_name = (
+            self.device_info["name"]
+            if self.device_info and "name" in self.device_info
+            else "New Device"
+        )
+        step_device_options_schema = vol.Schema(
+            {
+                vol.Required("custom_device_name", default=default_device_name): str,
+                vol.Required("polling_time", default=60): vol.All(
+                    vol.Coerce(int), vol.Clamp(min=60)
+                ),
+                vol.Required("group_sensors", default=True): bool,
+                vol.Required("disable_sensors", default=True): bool,
+            }
+        )
 
         # Show the form for step 2 with the device name and other options as defined in STEP_DEVICE_OPTIONS_SCHEMA
         return self.async_show_form(
@@ -169,8 +186,6 @@ async def async_step_device_options(
         )
 
 
-
-
 class CannotConnect(HomeAssistantError):
     """Error to indicate we cannot connect."""
 
@@ -179,23 +194,25 @@ class InvalidAuth(HomeAssistantError):
     """Error to indicate there is invalid auth."""
 
 
-#Helper function to sanitize
+# Helper function to sanitize
 def sanitize_device_name(device_name: str, fall_back: str, max_length=255) -> str:
     # Trim whitespace
     name = device_name.strip()
 
     # Remove special characters but keep spaces
-    name = re.sub(r'[^\w\s-]', '', name)
+    name = re.sub(r"[^\w\s-]", "", name)
 
     # Replace multiple spaces with a single space
-    name = re.sub(r'\s+', ' ', name)
+    name = re.sub(r"\s+", " ", name)
 
     # Length check
     if len(name) > max_length:
-        name = name[:max_length].rsplit(' ', 1)[0]  # Split at the last space to avoid cutting off in the middle of a word
+        name = name[:max_length].rsplit(" ", 1)[
+            0
+        ]  # Split at the last space to avoid cutting off in the middle of a word
 
     # Fallback name
     if not name:
         name = fall_back
 
-    return name
+    return name