Added reading of PortStatisticsRpm.htm

custom_components/tplink_easy_smart/binary_sensor.py

@@ -188,12 +188,22 @@ def _handle_coordinator_update(self) -> None:
             )
 
             self._attr_extra_state_attributes["number"] = port_info.number
+            self._attr_extra_state_attributes["enabled"] = 'Enabled' if port_info.enabled else 'Disabled'
+            self._attr_extra_state_attributes["flow_control_config"] = 'On' if port_info.flow_control_config else 'Off'
+            self._attr_extra_state_attributes["flow_control_actual"] = 'On' if port_info.flow_control_actual else 'Off'
             self._attr_extra_state_attributes["speed"] = DISPLAYED_PORT_SPEED.get(
                 port_info.speed_actual
             )
             self._attr_extra_state_attributes[
                 "speed_config"
             ] = DISPLAYED_PORT_SPEED.get(port_info.speed_config)
+
+            port_statistics= self.coordinator.get_port_statistics(self._port_number)
+            if port_statistics:
+                self._attr_extra_state_attributes["tx_good_packets"] = port_statistics.tx_good_pkts
+                self._attr_extra_state_attributes["rx_good_packets"] = port_statistics.rx_good_pkts
+                self._attr_extra_state_attributes["tx_bad_packets"] = port_statistics.tx_bad_pkts
+                self._attr_extra_state_attributes["rx_bad_packets"] = port_statistics.rx_bad_pkts
         else:
             self._attr_available = False
             self._attr_is_on = None

custom_components/tplink_easy_smart/client/classes.py

@@ -119,6 +119,19 @@ class PortState:
     speed_actual: PortSpeed
 
 
+# ---------------------------
+#   PortStatistics
+# ---------------------------
+@dataclass
+class PortStatistics:
+    number: int
+    enabled: bool
+    tx_good_pkts: int
+    tx_bad_pkts: int
+    rx_good_pkts: int
+    rx_bad_pkts: int
+
+
 # ---------------------------
 #   PortPoeState
 # ---------------------------

custom_components/tplink_easy_smart/client/const.py

@@ -3,9 +3,11 @@
 URL_DEVICE_INFO: Final = "SystemInfoRpm.htm"
 URL_PORTS_SETTINGS_GET: Final = "PortSettingRpm.htm"
 URL_POE_SETTINGS_GET: Final = "PoeConfigRpm.htm"
+URL_PORT_STATISTICS_GET: Final = "PortStatisticsRpm.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"
+FEATURE_STATS: Final = "feature_stats"

custom_components/tplink_easy_smart/client/tplink_api.py

@@ -13,15 +13,18 @@
     PortSpeed,
     PortState,
     TpLinkSystemInfo,
+    PortStatistics,
 )
 from .const import (
     FEATURE_POE,
+    FEATURE_STATS,
     URL_DEVICE_INFO,
     URL_POE_PORT_SETTINGS_SET,
     URL_POE_SETTINGS_GET,
     URL_POE_SETTINGS_SET,
     URL_PORT_SETTINGS_SET,
     URL_PORTS_SETTINGS_GET,
+    URL_PORT_STATISTICS_GET,
 )
 from .coreapi import TpLinkWebApi, VariableType
 from .utils import TpLinkFeaturesDetector
@@ -168,6 +171,47 @@ async def get_port_states(self) -> list[PortState]:
 
         return result
 
+    async def get_port_statistics(self) -> list[PortStatistics]:
+        """Return the port states."""
+        if not await self.is_feature_available(FEATURE_STATS):
+            return []
+        data = await self._core_api.get_variables(
+            URL_PORT_STATISTICS_GET,
+            [
+                ("all_info", VariableType.Dict),
+                ("max_port_num", VariableType.Int),
+            ],
+        )
+
+        result: list[PortStatistics] = []
+
+        all_info = data.get("all_info")
+        if not all_info:
+            return result
+
+        max_port_num = data.get("max_port_num")
+        if not max_port_num:
+            return result
+
+        pkts = all_info.get("pkts")
+        enabled_flags = all_info.get("state")
+        k = 0
+        for number in range(1, max_port_num + 1):
+            if k + 3 > len(pkts):
+                break
+            state = PortStatistics(
+                number=number,
+                enabled=enabled_flags[number - 1] == 1,
+                tx_good_pkts=pkts[k + 0],
+                tx_bad_pkts=pkts[k + 1],
+                rx_good_pkts=pkts[k + 2],
+                rx_bad_pkts=pkts[k + 3],
+            )
+            k += 4
+            result.append(state)
+
+        return result
+
     async def get_port_poe_states(self) -> list[PortPoeState]:
         """Return the port states."""
         if not await self.is_feature_available(FEATURE_POE):

custom_components/tplink_easy_smart/client/utils.py

@@ -1,7 +1,7 @@
 import logging
 from functools import wraps
 
-from .const import FEATURE_POE, URL_POE_SETTINGS_GET
+from .const import FEATURE_POE, URL_POE_SETTINGS_GET, FEATURE_STATS, URL_PORT_STATISTICS_GET
 from .coreapi import (
     ApiCallError,
     TpLinkWebApi,
@@ -70,10 +70,24 @@ async def _is_poe_available(self) -> bool:
         )
         return data.get("portConfig") is not None and data.get("poe_port_num") > 0
 
+    @log_feature(FEATURE_STATS)
+    @disconnected_as_false
+    async def _is_stats_available(self) -> bool:
+        data = await self._core_api.get_variables(
+            URL_PORT_STATISTICS_GET,
+            [
+                ("all_info", VariableType.Dict),
+                ("max_port_num", VariableType.Int),
+            ],
+        )
+        return data.get("all_info") is not None and data.get("max_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)
+        if await self._is_stats_available():
+            self._available_features.add(FEATURE_STATS)
 
     def is_available(self, feature: str) -> bool:
         """Return true if feature is available."""

custom_components/tplink_easy_smart/update_coordinator.py

@@ -18,8 +18,15 @@
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
 from .client.classes import PoePowerLimit, PoePriority, TpLinkSystemInfo
-from .client.const import FEATURE_POE
-from .client.tplink_api import PoeState, PortPoeState, PortSpeed, PortState, TpLinkApi
+from .client.const import FEATURE_POE, FEATURE_STATS
+from .client.tplink_api import (
+    PoeState,
+    PortPoeState,
+    PortSpeed,
+    PortState,
+    TpLinkApi,
+    PortStatistics,
+)
 from .const import ATTR_MANUFACTURER, DEFAULT_SCAN_INTERVAL, DOMAIN
 
 _LOGGER = logging.getLogger(__name__)
@@ -45,6 +52,7 @@ def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
         self._port_states: list[PortState] = []
         self._port_poe_states: list[PortPoeState] = []
         self._poe_state: PoeState | None = None
+        self._port_statistics: list[PortStatistics] = []
 
         update_interval = config_entry.options.get(
             CONF_SCAN_INTERVAL,
@@ -90,6 +98,16 @@ def get_port_state(self, number: int) -> PortState | None:
             return None
         return self._port_states[number - 1]
 
+    def get_port_statistics(self, number: int) -> PortStatistics | None:
+        """Return the specified port statistics."""
+        if (
+            number > self.ports_count
+            or number < 1
+            or len(self._port_statistics) < number
+        ):
+            return None
+        return self._port_statistics[number - 1]
+
     def get_port_poe_state(self, number: int) -> PortPoeState | None:
         """Return the specified port PoE state."""
         if number > self.ports_poe_count or number < 1:
@@ -122,6 +140,7 @@ async def async_update(self) -> None:
         await self._update_port_states()
         await self._update_poe_state()
         await self._update_port_poe_states()
+        await self._update_port_statistics()
         _LOGGER.debug("Update completed")
 
     def unload(self) -> None:
@@ -140,6 +159,17 @@ async def _update_port_states(self):
             _LOGGER.warning("Can not get port states: %s", repr(ex))
             self._port_states = []
 
+    async def _update_port_statistics(self):
+        """Update port statistis."""
+        if not await self.is_feature_available(FEATURE_STATS):
+            return
+
+        try:
+            self._port_statistics = await self._api.get_port_statistics()
+        except Exception as ex:
+            _LOGGER.warning("Can not get port statistics: %s", repr(ex))
+            self._port_statistics = []
+
     async def _update_poe_state(self):
         """Update the switch PoE state."""