Added auto-detection of PoE function availability

Author: vmakeev

README.md

@@ -24,13 +24,15 @@ Home Assistant custom component for control TP-Link Easy Smart switches over LAN
 - obtaining hardware and firmware version of the switch
 - obtaining information about total PoE consumption
 - setting limits on total PoE consumption
+- automatic detection of available functions
 
 ## Supported models
 
-|                                         Name                                          |  Revision | Confirmed |           Notes                         |
-|---------------------------------------------------------------------------------------|-----------|-----------|-----------------------------------------|
-| [TL-SG1016PE](https://www.tp-link.com/en/business-networking/poe-switch/tl-sg1016pe/) |   V1, V3  |    Yes    | All features are available              |
-| Other Easy Smart switches with web-based user interface                               | --------- |    No     | Will most likely work                   
+|                                          Name                                            |  Revision | Confirmed |           Notes                         |
+|------------------------------------------------------------------------------------------|-----------|-----------|-----------------------------------------|
+| [TL-SG1016PE](https://www.tp-link.com/en/business-networking/poe-switch/tl-sg1016pe/)    |   V1, V3  |    Yes    | All features are available              |
+| [TL-SG105E](https://www.tp-link.com/en/business-networking/easy-smart-switch/tl-sg105e/) |     V5    |    Yes    | PoE is not supported by device          |
+| Other Easy Smart switches with web-based user interface                                  | --------- |    No     | Will most likely work                   
 
 ## Installation
 

custom_components/tplink_easy_smart/client/const.py

@@ -0,0 +1,11 @@
+from typing import Final
+
+URL_DEVICE_INFO: Final = "SystemInfoRpm.htm"
+URL_PORTS_SETTINGS_GET: Final = "PortSettingRpm.htm"
+URL_POE_SETTINGS_GET: Final = "PoeConfigRpm.htm"
+
+URL_PORT_SETTINGS_SET: Final = "port_setting.cgi"
+URL_POE_SETTINGS_SET: Final = "poe_global_config.cgi"
+URL_POE_PORT_SETTINGS_SET: Final = "poe_port_config.cgi"
+
+FEATURE_POE: Final = "feature_poe"

custom_components/tplink_easy_smart/client/coreapi.py

@@ -1,23 +1,25 @@
 """TP-Link web api core functions."""
 
 import asyncio
-from enum import Enum
 import logging
 import re
+from enum import Enum
 from typing import Callable, Dict, Final, Iterable, Tuple, TypeAlias
 
 import aiohttp
-from aiohttp import ClientResponse
 import json5
+from aiohttp import ClientResponse, ServerDisconnectedError
 
 TIMEOUT: Final = 5.0
 
 APICALL_ERRCODE_UNAUTHORIZED: Final = -2
 APICALL_ERRCODE_REQUEST: Final = -3
+APICALL_ERRCODE_DISCONNECTED: Final = -4
 
 APICALL_ERRCAT_CREDENTIALS: Final = "user_pass_err"
 APICALL_ERRCAT_REQUEST: Final = "request_error"
 APICALL_ERRCAT_UNAUTHORIZED: Final = "unauthorized"
+APICALL_ERRCAT_DISCONNECTED: Final = "disconnected"
 
 AUTH_FAILURE_GENERAL: Final = "auth_general"
 AUTH_FAILURE_CREDENTIALS: Final = "auth_invalid_credentials"
@@ -249,6 +251,12 @@ async def _get_raw(self, path: str) -> ClientResponse:
             )
             _LOGGER.debug("GET %s performed, status: %s", path, response.status)
             return response
+        except ServerDisconnectedError as sde:
+            raise ApiCallError(
+                f"Can not perform GET request at {path} cause of {repr(sde)}",
+                APICALL_ERRCODE_DISCONNECTED,
+                APICALL_ERRCAT_DISCONNECTED,
+            )
         except Exception as ex:
             _LOGGER.error("GET %s failed: %s", path, str(ex))
             raise ApiCallError(
@@ -269,6 +277,12 @@ async def _post_raw(self, path: str, data: Dict) -> ClientResponse:
             )
             _LOGGER.debug("POST to %s performed, status: %s", path, response.status)
             return response
+        except ServerDisconnectedError as sde:
+            raise ApiCallError(
+                f"Can not perform POST request at {path} cause of {repr(sde)}",
+                APICALL_ERRCODE_DISCONNECTED,
+                APICALL_ERRCAT_DISCONNECTED,
+            )
         except Exception as ex:
             _LOGGER.error("POST %s failed: %s", path, str(ex))
             raise ApiCallError(

custom_components/tplink_easy_smart/client/tplink_api.py

@@ -1,7 +1,7 @@
 """TP-Link api."""
 
 import logging
-from typing import Final, Tuple
+from typing import Tuple
 
 from .classes import (
     PoeClass,
@@ -14,18 +14,20 @@
     PortState,
     TpLinkSystemInfo,
 )
+from .const import (
+    FEATURE_POE,
+    URL_DEVICE_INFO,
+    URL_POE_PORT_SETTINGS_SET,
+    URL_POE_SETTINGS_GET,
+    URL_POE_SETTINGS_SET,
+    URL_PORT_SETTINGS_SET,
+    URL_PORTS_SETTINGS_GET,
+)
 from .coreapi import TpLinkWebApi, VariableType
+from .utils import TpLinkFeaturesDetector
 
 _LOGGER = logging.getLogger(__name__)
 
-_URL_DEVICE_INFO: Final = "SystemInfoRpm.htm"
-_URL_PORTS_SETTINGS_GET: Final = "PortSettingRpm.htm"
-_URL_POE_SETTINGS_GET: Final = "PoeConfigRpm.htm"
-
-_URL_PORT_SETTINGS_SET: Final = "port_setting.cgi"
-_URL_POE_SETTINGS_SET: Final = "poe_global_config.cgi"
-_URL_POE_PORT_SETTINGS_SET: Final = "poe_port_config.cgi"
-
 _POE_PRIORITIES_SET_MAP: dict[PoePriority, int] = {
     PoePriority.HIGH: 1,
     PoePriority.MIDDLE: 2,
@@ -74,8 +76,22 @@ def __init__(
     ) -> None:
         """Initialize."""
         self._core_api = TpLinkWebApi(host, port, use_ssl, user, password, verify_ssl)
+        self._is_features_updated = False
+        self._features = TpLinkFeaturesDetector(self._core_api)
         _LOGGER.debug("New instance of TpLinkApi created")
 
+    async def _ensure_features_updated(self):
+        if not self._is_features_updated:
+            _LOGGER.debug("Updating available features")
+            await self._features.update()
+            self._is_features_updated = True
+            _LOGGER.debug("Available features updated")
+
+    async def is_feature_available(self, feature: str) -> bool:
+        """Return true if specified feature is known and available."""
+        await self._ensure_features_updated()
+        return self._features.is_available(feature)
+
     async def authenticate(self) -> None:
         """Perform authentication."""
         await self._core_api.authenticate()
@@ -92,7 +108,7 @@ def device_url(self) -> str:
     async def get_device_info(self) -> TpLinkSystemInfo:
         """Return the device information."""
         data = await self._core_api.get_variable(
-            _URL_DEVICE_INFO, "info_ds", VariableType.Dict
+            URL_DEVICE_INFO, "info_ds", VariableType.Dict
         )
 
         def get_value(key: str) -> str | None:
@@ -116,7 +132,7 @@ def get_value(key: str) -> str | None:
     async def get_port_states(self) -> list[PortState]:
         """Return the port states."""
         data = await self._core_api.get_variables(
-            _URL_PORTS_SETTINGS_GET,
+            URL_PORTS_SETTINGS_GET,
             [
                 ("all_info", VariableType.Dict),
                 ("max_port_num", VariableType.Int),
@@ -154,8 +170,11 @@ async def get_port_states(self) -> list[PortState]:
 
     async def get_port_poe_states(self) -> list[PortPoeState]:
         """Return the port states."""
+        if not await self.is_feature_available(FEATURE_POE):
+            return []
+
         data = await self._core_api.get_variables(
-            _URL_POE_SETTINGS_GET,
+            URL_POE_SETTINGS_GET,
             [
                 ("portConfig", VariableType.Dict),
                 ("poe_port_num", VariableType.Int),
@@ -202,11 +221,13 @@ async def get_port_poe_states(self) -> list[PortPoeState]:
 
     async def get_poe_state(self) -> PoeState | None:
         """Return the port states."""
+        if not await self.is_feature_available(FEATURE_POE):
+            return None
 
         _LOGGER.debug("Begin fetching POE states")
 
         poe_config = await self._core_api.get_variable(
-            _URL_POE_SETTINGS_GET, "globalConfig", VariableType.Dict
+            URL_POE_SETTINGS_GET, "globalConfig", VariableType.Dict
         )
         if not poe_config:
             _LOGGER.debug("No globalConfig found, returning")
@@ -235,10 +256,13 @@ async def set_port_state(
             f"flowcontrol={1 if flow_control_config else 0}&"
             f"apply=Apply"
         )
-        await self._core_api.get(_URL_PORT_SETTINGS_SET, query=query)
+        await self._core_api.get(URL_PORT_SETTINGS_SET, query=query)
 
     async def set_poe_limit(self, limit: float) -> None:
         """Change poe limit."""
+        if not await self.is_feature_available(FEATURE_POE):
+            raise ActionError("POE feature is not supported by device")
+
         current_state = await self.get_poe_state()
         if not current_state:
             raise ActionError("Can not get actual PoE state")
@@ -258,7 +282,7 @@ async def set_poe_limit(self, limit: float) -> None:
             "name_powerremain": current_state.power_remain,
             "applay": "Apply",
         }
-        result = await self._core_api.post(_URL_POE_SETTINGS_SET, data)
+        result = await self._core_api.post(URL_POE_SETTINGS_SET, data)
         _LOGGER.debug("POE_SET_RESULT: %s", result)
 
     async def set_port_poe_settings(
@@ -268,12 +292,14 @@ async def set_port_poe_settings(
         priority: PoePriority,
         power_limit: PoePowerLimit | float,
     ) -> None:
+        if not await self.is_feature_available(FEATURE_POE):
+            raise ActionError("POE feature is not supported by device")
         """Change port poe settings."""
         if port_number < 1:
             raise ActionError("Port number should be greater than or equals to 1")
 
         poe_ports_count = await self._core_api.get_variable(
-            _URL_POE_SETTINGS_GET, "poe_port_num", VariableType.Int
+            URL_POE_SETTINGS_GET, "poe_port_num", VariableType.Int
         )
         if not poe_ports_count:
             raise ActionError("Can not get PoE ports count")
@@ -310,5 +336,5 @@ async def set_port_poe_settings(
             f"sel_{port_number}": 1,
             "applay": "Apply",
         }
-        result = await self._core_api.post(_URL_POE_PORT_SETTINGS_SET, data)
+        result = await self._core_api.post(URL_POE_PORT_SETTINGS_SET, data)
         _LOGGER.debug("POE_PORT_SETTINGS_SET_RESULT: %s", result)

custom_components/tplink_easy_smart/client/utils.py

@@ -0,0 +1,80 @@
+import logging
+from functools import wraps
+
+from .const import FEATURE_POE, URL_POE_SETTINGS_GET
+from .coreapi import (
+    ApiCallError,
+    TpLinkWebApi,
+    VariableType,
+    APICALL_ERRCAT_DISCONNECTED,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+# ---------------------------
+#   TpLinkFeaturesDetector
+# ---------------------------
+class TpLinkFeaturesDetector:
+    def __init__(self, core_api: TpLinkWebApi):
+        """Initialize."""
+        self._core_api = core_api
+        self._available_features = set()
+        self._is_initialized = False
+
+    @staticmethod
+    def disconnected_as_false(func):
+        @wraps(func)
+        async def wrapper(*args, **kwargs) -> bool:
+            try:
+                return await func(*args, **kwargs)
+            except ApiCallError as ace:
+                if ace.category == APICALL_ERRCAT_DISCONNECTED:
+                    return False
+                raise
+
+        return wrapper
+
+    @staticmethod
+    def log_feature(feature_name: str):
+        def decorator(func):
+            @wraps(func)
+            async def wrapper(*args, **kwargs):
+                try:
+                    _LOGGER.debug("Check feature '%s' availability", feature_name)
+                    result = await func(*args, **kwargs)
+                    if result:
+                        _LOGGER.debug("Feature '%s' is available", feature_name)
+                    else:
+                        _LOGGER.debug("Feature '%s' is not available", feature_name)
+                    return result
+                except Exception:
+                    _LOGGER.debug(
+                        "Feature availability check failed on %s", feature_name
+                    )
+                    raise
+
+            return wrapper
+
+        return decorator
+
+    @log_feature(FEATURE_POE)
+    @disconnected_as_false
+    async def _is_poe_available(self) -> bool:
+        data = await self._core_api.get_variables(
+            URL_POE_SETTINGS_GET,
+            [
+                ("portConfig", VariableType.Dict),
+                ("poe_port_num", VariableType.Int),
+            ],
+        )
+        return data.get("portConfig") is not None and data.get("poe_port_num") > 0
+
+    async def update(self) -> None:
+        """Update the available features list."""
+        if await self._is_poe_available():
+            self._available_features.add(FEATURE_POE)
+
+    def is_available(self, feature: str) -> bool:
+        """Return true if feature is available."""
+        return feature in self._available_features

custom_components/tplink_easy_smart/sensor.py

@@ -15,8 +15,8 @@
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
+from .client.const import FEATURE_POE
 
-from .const import DATA_KEY_COORDINATOR, DOMAIN
 from .helpers import (
     generate_entity_id,
     generate_entity_name,
@@ -74,21 +74,25 @@ async def async_setup_entry(
                 function_name=_FUNCTION_DISPLAYED_NAME_NETWORK_INFO,
             ),
         ),
-        TpLinkPoeInfoSensor(
-            coordinator,
-            TpLinkSensorEntityDescription(
-                key="poe_consumption",
-                icon="mdi:lightning-bolt",
-                device_class=SensorDeviceClass.POWER,
-                native_unit_of_measurement=POWER_WATT,
-                state_class=SensorStateClass.MEASUREMENT,
-                device_name=coordinator.get_switch_info().name,
-                function_uid=_FUNCTION_UID_POE_INFO,
-                function_name=_FUNCTION_DISPLAYED_NAME_POE_INFO,
-            ),
-        ),
     ]
 
+    if await coordinator.is_feature_available(FEATURE_POE):
+        sensors.append(
+            TpLinkPoeInfoSensor(
+                coordinator,
+                TpLinkSensorEntityDescription(
+                    key="poe_consumption",
+                    icon="mdi:lightning-bolt",
+                    device_class=SensorDeviceClass.POWER,
+                    native_unit_of_measurement=POWER_WATT,
+                    state_class=SensorStateClass.MEASUREMENT,
+                    device_name=coordinator.get_switch_info().name,
+                    function_uid=_FUNCTION_UID_POE_INFO,
+                    function_name=_FUNCTION_DISPLAYED_NAME_POE_INFO,
+                ),
+            )
+        )
+
     async_add_entities(sensors)