Fixed #53 - integration fails to load EdgeOS older than 2.0.9

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## v1.1.5
+- Fixed integration fails to load with EdgeOS version older than 2.0.9 [\#53](https://github.com/elad-bar/ha-edgeos/pull/53)
+
 ## v1.1.4
 - Prevent the component to get installed or run with EdgeOS Firmware v1
 

custom_components/edgeos/clients/__init__.py

@@ -1,14 +0,0 @@
-from homeassistant.exceptions import HomeAssistantError
-
-
-class SessionTerminatedException(HomeAssistantError):
-    Terminated = True
-
-
-class LoginException(HomeAssistantError):
-    def __init__(self, status_code):
-        self._status_code = status_code
-
-    @property
-    def status_code(self):
-        return self._status_code

custom_components/edgeos/clients/web_api.py

@@ -12,9 +12,9 @@
 
 from homeassistant.helpers.aiohttp_client import async_create_clientsession
 
-from . import LoginException, SessionTerminatedException
 from ..helpers.const import *
 from ..managers.configuration_manager import ConfigManager
+from ..models.exceptions import LoginException, SessionTerminatedException
 from .web_socket import EdgeOSWebSocket
 
 REQUIREMENTS = ["aiohttp"]

custom_components/edgeos/helpers/const.py

@@ -34,6 +34,9 @@
     DOMAIN_DEVICE_TRACKER: SIGNAL_UPDATE_TRACKERS,
 }
 
+EDGEOS_VERSION_INCOMPATIBLE = "v1"
+EDGEOS_VERSION_UNKNOWN = "N/A"
+
 MANUFACTURER = "Ubiquiti"
 
 NOTIFICATION_ID = "edgeos_notification"

custom_components/edgeos/managers/config_flow_manager.py

@@ -5,14 +5,15 @@
 
 from homeassistant.config_entries import ConfigEntry
 
-from .. import get_ha
-from ..clients import LoginException
 from ..clients.web_api import EdgeOSWebAPI
+from ..helpers import get_ha
 from ..helpers.const import *
 from ..managers.configuration_manager import ConfigManager
 from ..managers.password_manager import PasswordManager
 from ..models import AlreadyExistsError, LoginError
 from ..models.config_data import ConfigData
+from ..models.exceptions import IncompatibleVersion, LoginException
+from .version_check import VersionManager
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -133,7 +134,7 @@ def _validate_unique_name(self, user_input):
             if ha is not None:
                 raise AlreadyExistsError(entry_primary_key)
 
-    def _get_default_fields(self, flow, config_data: Optional[ConfigData] = None):
+    def _get_default_fields(self, flow, config_data: Optional[ConfigData] = None) -> dict:
         if config_data is None:
             config_data = self.config_data
 
@@ -226,19 +227,25 @@ def get_default_options(self):
 
     async def _update_entry(self):
         try:
-            entry = ConfigEntry(
-                0, "", "", self._data, "", "", {}, options=self._options
-            )
+            entry = ConfigEntry(version=0,
+                                domain="",
+                                title="",
+                                data=self._data,
+                                source="",
+                                options=self._options)
 
             await self._config_manager.update(entry)
         except InvalidToken:
             _LOGGER.info("Reset password")
 
             del self._data[CONF_PASSWORD]
 
-            entry = ConfigEntry(
-                0, "", "", self._data, "", "", {}, options=self._options
-            )
+            entry = ConfigEntry(version=0,
+                                domain="",
+                                title="",
+                                data=self._data,
+                                source="",
+                                options=self._options)
 
             await self._config_manager.update(entry)
 
@@ -421,17 +428,9 @@ async def _valid_login(self):
                         errors = {"base": "invalid_export_configuration"}
 
                     system_info_data = await api.get_general_data(SYS_INFO_KEY)
-
-                    if system_info_data is not None:
-                        firmware_version = system_info_data.get("fw-latest", {})
-                        version = firmware_version.get("version")
-
-                        if version[:2] == "v1":
-                            _LOGGER.error(
-                                f"Unsupported firmware version ({version})"
-                            )
-
-                            errors = {"base": "incompatible_version"}
+                    vm = VersionManager()
+                    vm.update(system_info_data)
+                    vm.validate()
 
             else:
                 _LOGGER.warning(f"Failed to login {name}")
@@ -445,6 +444,11 @@ async def _valid_login(self):
 
             errors = {"base": HTTP_ERRORS.get(ex.status_code, "auth_general_error")}
 
+        except IncompatibleVersion as ivex:
+            _LOGGER.error(str(ivex))
+
+            errors = {"base": "incompatible_version"}
+
         except Exception as ex:
             _LOGGER.warning(f"Failed to login {name} due to general error: {str(ex)}")
 

custom_components/edgeos/managers/data_manager.py

@@ -8,13 +8,13 @@
 import sys
 from typing import Optional
 
-from custom_components.edgeos.clients import SessionTerminatedException
-
 from ..clients.web_api import EdgeOSWebAPI
 from ..clients.web_socket import EdgeOSWebSocket
 from ..helpers.const import *
 from ..models.config_data import ConfigData
+from ..models.exceptions import IncompatibleVersion, SessionTerminatedException
 from .configuration_manager import ConfigManager
+from .version_check import VersionManager
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -24,6 +24,7 @@ class EdgeOSData:
     version: str
     edgeos_data: dict
     system_data: dict
+    version_manager: VersionManager
 
     def __init__(self, hass, config_manager: ConfigManager, update_home_assistant):
         self._hass = hass
@@ -43,16 +44,21 @@ def __init__(self, hass, config_manager: ConfigManager, update_home_assistant):
         topics = self._ws_handlers.keys()
 
         self.hostname = config_data.host
-        self.version = "N/A"
 
         self._ws = EdgeOSWebSocket(self._hass, config_manager, topics, self.ws_handler)
 
         self._api = EdgeOSWebAPI(
             self._hass, config_manager, self.edgeos_disconnection_handler, self._ws
         )
 
+        self.version_manager = VersionManager()
+
         self._is_active = True
 
+    @property
+    def version(self):
+        return self.version_manager.version
+
     @property
     def product(self):
         return self._api.product
@@ -119,22 +125,21 @@ async def _initialize(self, post_login_action=None):
                 cookies = self._api.cookies_data
                 session_id = self._api.session_id
 
-                if self.version[:2] == "v1":
-                    _LOGGER.error(
-                        f"Unsupported firmware version ({self.version})"
-                    )
-
-                    await self.terminate()
+                self.version_manager.validate()
 
-                else:
-                    _LOGGER.debug(f"Initializing WS using session: {session_id}")
-                    await self._ws.initialize(cookies, session_id)
+                _LOGGER.debug(f"Initializing WS using session: {session_id}")
+                await self._ws.initialize(cookies, session_id)
 
         except SessionTerminatedException as stex:
             _LOGGER.info(f"Session terminated ({stex})")
 
             self._is_active = False
 
+        except IncompatibleVersion as ivex:
+            _LOGGER.error(str(ivex))
+
+            self._is_active = False
+
         except Exception as ex:
             exc_type, exc_obj, tb = sys.exc_info()
             line_number = tb.tb_lineno
@@ -269,7 +274,7 @@ def ws_handler(self, payload=None):
                 f"Failed to handle WS message, Error: {ex}, Line: {line_number}"
             )
 
-    def get_ws_handlers(self):
+    def get_ws_handlers(self) -> dict:
         ws_handlers = {
             EXPORT_KEY: self.handle_export,
             INTERFACES_KEY: self.handle_interfaces,
@@ -373,11 +378,8 @@ def load_system_data(self, devices_data, system_info_data):
             return
 
         system_data = devices_data.get("system", {})
-        firmware_version = system_info_data.get("fw-latest", {})
-        version = firmware_version.get("version")
-
         self.hostname = system_data.get("host-name", self.hostname)
-        self.version = version
+        self.version_manager.update(system_info_data)
 
     def handle_interfaces(self, data):
         try:

custom_components/edgeos/managers/version_check.py

@@ -1,40 +1,39 @@
 import logging
+from typing import Optional
+
+from ..helpers.const import *
+from ..models.exceptions import IncompatibleVersion
 
 _LOGGER = logging.getLogger(__name__)
 
-MAX_PARTS = 3
-MINIMUM_EDGEOS_VERSION = "1.10"
 
+class VersionManager:
+    version: Optional[str]
 
-class VersionCheck:
     def __init__(self):
-        self._minimum_version_score = self._get_score(MINIMUM_EDGEOS_VERSION)
-
-    @staticmethod
-    def _get_score(version):
-        ver_number_arr = version.split(".")
-
-        multiplier = 100000000
-        total_version = 0
-
-        max_numbers_to_check = len(ver_number_arr)
-        if max_numbers_to_check > MAX_PARTS:
-            max_numbers_to_check = MAX_PARTS
+        self.version = None
 
-        for i in range(0, max_numbers_to_check):
-            ver_id = int(ver_number_arr[i]) * multiplier
-            total_version = total_version + ver_id
+    def validate(self):
+        if self.version is None:
+            raise IncompatibleVersion(self.version)
 
-            multiplier = multiplier / 10000
+        if self.version == EDGEOS_VERSION_UNKNOWN:
+            raise IncompatibleVersion(self.version)
 
-        return total_version
+        if self.version[:2] == EDGEOS_VERSION_INCOMPATIBLE:
+            raise IncompatibleVersion(self.version)
 
-    def is_compatible(self, fw_version):
-        version_parts = fw_version.split("v")
-        version = version_parts[len(version_parts) - 1]
+    def update(self, system_info_data):
+        firmware_version = system_info_data.get("fw-latest", None)
+        if firmware_version is None:
+            software_version = system_info_data.get("sw_ver", "N/A")
+            software_version_items = software_version.split(".")
 
-        version_score = self._get_score(version)
+            version_without_build = software_version_items[:-3]
+            version_without_model = version_without_build[2:]
+            version = '.'.join(version_without_model)
 
-        is_compatible = version_score >= self._minimum_version_score
+        else:
+            version = firmware_version.get("version", "N/A")
 
-        return is_compatible
+        self.version = version

custom_components/edgeos/manifest.json

@@ -6,6 +6,6 @@
   "codeowners": ["@elad-bar"],
   "requirements": ["aiohttp"],
   "config_flow": true,
-  "version": "1.1.4",
+  "version": "1.1.5",
   "iot_class": "local_polling"
 }

custom_components/edgeos/models/exceptions.py

@@ -0,0 +1,22 @@
+from homeassistant.exceptions import HomeAssistantError
+
+
+class IncompatibleVersion(HomeAssistantError):
+    def __init__(self, version):
+        self._version = version
+
+    def __repr__(self):
+        return f"Unsupported EdgeOS version ({self._version})"
+
+
+class SessionTerminatedException(HomeAssistantError):
+    Terminated = True
+
+
+class LoginException(HomeAssistantError):
+    def __init__(self, status_code):
+        self._status_code = status_code
+
+    @property
+    def status_code(self):
+        return self._status_code