Merge pull request #143 from elad-bar/rename-status-entities

Fix status entities based on is admin validation

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 - Update supported interfaces [#137](https://github.com/elad-bar/ha-edgeos/issues/137)
 - Removed unused constants
+- Fix status entities, if EdgeOS user is admin, the switch will be available, if not, binary sensor
+- Adjusted the code of entity validations (whether) to present or not
 
 ## 2.1.2
 

custom_components/edgeos/common/base_entity.py

@@ -34,19 +34,11 @@ def _async_handle_device(
         try:
             coordinator = hass.data[DOMAIN][entry.entry_id]
 
-            if device_type == DeviceTypes.DEVICE:
-                is_monitored = coordinator.config_manager.get_monitored_device(item_id)
-
-            elif device_type == DeviceTypes.INTERFACE:
-                is_monitored = coordinator.config_manager.get_monitored_interface(
-                    item_id
-                )
-
-            else:
-                is_monitored = True
+            is_admin = coordinator.system.is_admin
+            is_monitored = coordinator.config_manager.is_monitored(device_type, item_id)
 
             entity_descriptions = get_entity_descriptions(
-                platform, device_type, is_monitored
+                platform, device_type, is_monitored, is_admin
             )
 
             entities = [
@@ -134,6 +126,22 @@ def _local_coordinator(self) -> Coordinator:
     def data(self) -> dict | None:
         return self._data
 
+    @property
+    def _is_allowed_for_monitoring(self) -> bool:
+        is_allowed_for_monitoring = False
+
+        if self._device_type == DeviceTypes.DEVICE:
+            is_allowed_for_monitoring = (
+                self.coordinator.config_manager.get_monitored_interface(self._item_id)
+            )
+
+        if self._device_type == DeviceTypes.DEVICE:
+            is_allowed_for_monitoring = (
+                self.coordinator.config_manager.get_monitored_device(self._item_id)
+            )
+
+        return is_allowed_for_monitoring
+
     async def async_execute_device_action(self, key: str, *kwargs: Any):
         async_device_action = self._local_coordinator.get_device_action(
             self._entity_description, self._item_id, key

custom_components/edgeos/common/consts.py

@@ -7,7 +7,13 @@
 
 from homeassistant.const import UnitOfDataRate, UnitOfInformation
 
-from .enums import DeviceTypes, DynamicInterfaceTypes, InterfaceTypes, UnitOfEdgeOS
+from .enums import (
+    DeviceTypes,
+    DynamicInterfaceTypes,
+    EntityValidation,
+    InterfaceTypes,
+    UnitOfEdgeOS,
+)
 
 ENTITY_CONFIG_ENTRY_ID = "entry_id"
 
@@ -322,3 +328,9 @@
 ALL_EDGE_OS_UNITS = [str(unit) for unit in list(UnitOfEdgeOS)]
 
 SUPPORTED_REMOVED_ENTITIES_DEVICE_TYPES = [DeviceTypes.DEVICE, DeviceTypes.INTERFACE]
+
+ENTITY_VALIDATIONS = {
+    EntityValidation.MONITORED: lambda is_monitored, is_admin: is_monitored,
+    EntityValidation.ADMIN_ONLY: lambda is_monitored, is_admin: is_admin,
+    EntityValidation.NON_ADMIN_ONLY: lambda is_monitored, is_admin: not is_admin,
+}

custom_components/edgeos/common/entity_descriptions.py

@@ -1,9 +1,13 @@
 from copy import copy
 from dataclasses import dataclass
-from typing import Callable
 
-from custom_components.edgeos.common.consts import UNIT_MAPPING
-from custom_components.edgeos.common.enums import DeviceTypes, EntityKeys, UnitOfEdgeOS
+from custom_components.edgeos.common.consts import ENTITY_VALIDATIONS, UNIT_MAPPING
+from custom_components.edgeos.common.enums import (
+    DeviceTypes,
+    EntityKeys,
+    EntityValidation,
+    UnitOfEdgeOS,
+)
 from homeassistant.components.binary_sensor import (
     BinarySensorDeviceClass,
     BinarySensorEntityDescription,
@@ -24,7 +28,7 @@
 class IntegrationEntityDescription(EntityDescription):
     platform: Platform | None = None
     device_type: DeviceTypes | None = None
-    filter: Callable[[bool], bool] | None = lambda is_monitored: True
+    entity_validation: EntityValidation | None = None
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -142,88 +146,87 @@ class IntegrationNumberEntityDescription(
     IntegrationBinarySensorEntityDescription(
         key=EntityKeys.INTERFACE_CONNECTED,
         device_class=BinarySensorDeviceClass.CONNECTIVITY,
-        filter=lambda is_monitored: is_monitored,
         device_type=DeviceTypes.INTERFACE,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.INTERFACE_RECEIVED_DROPPED,
         native_unit_of_measurement=UnitOfEdgeOS.DROPPED,
         state_class=SensorStateClass.MEASUREMENT,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:package-variant-minus",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.INTERFACE_SENT_DROPPED,
         native_unit_of_measurement=UnitOfEdgeOS.DROPPED,
         state_class=SensorStateClass.MEASUREMENT,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:package-variant-minus",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.INTERFACE_RECEIVED_ERRORS,
         native_unit_of_measurement=UnitOfEdgeOS.ERRORS,
         state_class=SensorStateClass.MEASUREMENT,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:timeline-alert",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.INTERFACE_SENT_ERRORS,
         native_unit_of_measurement=UnitOfEdgeOS.ERRORS,
         state_class=SensorStateClass.MEASUREMENT,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:timeline-alert",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.INTERFACE_RECEIVED_PACKETS,
         native_unit_of_measurement=UnitOfEdgeOS.PACKETS,
         state_class=SensorStateClass.MEASUREMENT,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:package-up",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.INTERFACE_SENT_PACKETS,
         native_unit_of_measurement=UnitOfEdgeOS.PACKETS,
         state_class=SensorStateClass.MEASUREMENT,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:package-up",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.INTERFACE_RECEIVED_RATE,
         device_class=SensorDeviceClass.DATA_RATE,
         state_class=SensorStateClass.MEASUREMENT,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:download-network-outline",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.INTERFACE_SENT_RATE,
         device_class=SensorDeviceClass.DATA_RATE,
         state_class=SensorStateClass.MEASUREMENT,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:upload-network-outline",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.INTERFACE_RECEIVED_TRAFFIC,
         device_class=SensorDeviceClass.DATA_SIZE,
         state_class=SensorStateClass.TOTAL_INCREASING,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:download-network-outline",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.INTERFACE_SENT_TRAFFIC,
         device_class=SensorDeviceClass.DATA_SIZE,
         state_class=SensorStateClass.TOTAL_INCREASING,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:upload-network-outline",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSwitchEntityDescription(
         key=EntityKeys.INTERFACE_MONITORED,
@@ -235,48 +238,50 @@ class IntegrationNumberEntityDescription(
         key=EntityKeys.INTERFACE_STATUS,
         icon="mdi:monitor-eye",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.NON_ADMIN_ONLY,
     ),
     IntegrationSwitchEntityDescription(
         key=EntityKeys.INTERFACE_STATUS,
-        icon="mdi:monitor-eye",
+        icon="mdi:ethernet",
         device_type=DeviceTypes.INTERFACE,
+        entity_validation=EntityValidation.ADMIN_ONLY,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.DEVICE_RECEIVED_RATE,
         device_class=SensorDeviceClass.DATA_RATE,
         state_class=SensorStateClass.MEASUREMENT,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:download-network-outline",
         device_type=DeviceTypes.DEVICE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.DEVICE_SENT_RATE,
         device_class=SensorDeviceClass.DATA_RATE,
         state_class=SensorStateClass.MEASUREMENT,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:upload-network-outline",
         device_type=DeviceTypes.DEVICE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.DEVICE_RECEIVED_TRAFFIC,
         device_class=SensorDeviceClass.DATA_SIZE,
         state_class=SensorStateClass.TOTAL_INCREASING,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:download-network-outline",
         device_type=DeviceTypes.DEVICE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSensorEntityDescription(
         key=EntityKeys.DEVICE_SENT_TRAFFIC,
         device_class=SensorDeviceClass.DATA_SIZE,
         state_class=SensorStateClass.TOTAL_INCREASING,
-        filter=lambda is_monitored: is_monitored,
         icon="mdi:upload-network-outline",
         device_type=DeviceTypes.DEVICE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationDeviceTrackerEntityDescription(
         key=EntityKeys.DEVICE_TRACKER,
-        filter=lambda is_monitored: is_monitored,
         device_type=DeviceTypes.DEVICE,
+        entity_validation=EntityValidation.MONITORED,
     ),
     IntegrationSwitchEntityDescription(
         key=EntityKeys.DEVICE_MONITORED,
@@ -288,7 +293,10 @@ class IntegrationNumberEntityDescription(
 
 
 def get_entity_descriptions(
-    platform: Platform, device_type: DeviceTypes, is_monitored: bool
+    platform: Platform,
+    device_type: DeviceTypes,
+    is_monitored: bool | None,
+    is_admin: bool | None,
 ) -> list[IntegrationEntityDescription]:
     entity_descriptions = copy(ENTITY_DESCRIPTIONS)
 
@@ -297,12 +305,26 @@ def get_entity_descriptions(
         for entity_description in entity_descriptions
         if entity_description.platform == platform
         and entity_description.device_type == device_type
-        and entity_description.filter(is_monitored)
+        and is_valid_entity(entity_description, is_monitored, is_admin)
     ]
 
     return result
 
 
+def is_valid_entity(
+    entity_description: IntegrationEntityDescription,
+    is_monitored: bool | None,
+    is_admin: bool | None,
+):
+    is_valid = entity_description.entity_validation is None
+
+    if entity_description.entity_validation is not None:
+        validation = ENTITY_VALIDATIONS[entity_description.entity_validation]
+        is_valid = validation(is_monitored, is_admin)
+
+    return is_valid
+
+
 def get_platforms() -> list[str]:
     platforms = {
         entity_description.platform: None for entity_description in ENTITY_DESCRIPTIONS

custom_components/edgeos/common/enums.py

@@ -65,3 +65,9 @@ class UnitOfEdgeOS(StrEnum):
     DROPPED = "Dropped"
     PACKETS = "Packets"
     DEVICES = "Devices"
+
+
+class EntityValidation(StrEnum):
+    MONITORED = "monitored"
+    ADMIN_ONLY = "admin-only"
+    NON_ADMIN_ONLY = "non-admin-only"

custom_components/edgeos/data_processors/system_processor.py

@@ -137,30 +137,10 @@ def _process_api_data(self):
 
             self._system = system_data
 
-            message = (
-                f"User {self._config_data.username} level is {self._system.user_level}, "
-                f"Interface status switch will not be created as it requires admin role"
-            )
-
-            self._unique_log(logging.INFO, message)
-
             self._update_leased_devices()
 
-            warning_messages = []
-
-            if not self._system.deep_packet_inspection:
-                warning_messages.append("DPI (deep packet inspection) is turned off")
-
-            if not self._system.traffic_analysis_export:
-                warning_messages.append("Traffic Analysis Export is turned off")
-
-            if len(warning_messages) > 0:
-                warning_message = " and ".join(warning_messages)
-
-                self._unique_log(
-                    logging.WARNING,
-                    f"Integration will not work correctly since {warning_message}",
-                )
+            self._validate_admin()
+            self._validate_unit_settings()
 
         except Exception as ex:
             exc_type, exc_obj, tb = sys.exc_info()
@@ -220,6 +200,50 @@ def _update_leased_devices(self):
                 f"Failed to extract Unknown Devices data, Error: {ex}, Line: {line_number}"
             )
 
+    def _validate_admin(self):
+        try:
+            if not self._system.is_admin:
+                message = (
+                    f"User {self._config_data.username} level is {self._system.user_level}, "
+                    f"Interface status switch will not be created as it requires admin role"
+                )
+
+                self._unique_log(logging.INFO, message)
+
+        except Exception as ex:
+            exc_type, exc_obj, tb = sys.exc_info()
+            line_number = tb.tb_lineno
+
+            _LOGGER.error(
+                f"Failed to validate if user is admin, Error: {ex}, Line: {line_number}"
+            )
+
+    def _validate_unit_settings(self):
+        try:
+            warning_messages = []
+
+            if not self._system.deep_packet_inspection:
+                warning_messages.append("DPI (deep packet inspection) is turned off")
+
+            if not self._system.traffic_analysis_export:
+                warning_messages.append("Traffic Analysis Export is turned off")
+
+            if len(warning_messages) > 0:
+                warning_message = " and ".join(warning_messages)
+
+                self._unique_log(
+                    logging.WARNING,
+                    f"Integration will not work correctly since {warning_message}",
+                )
+
+        except Exception as ex:
+            exc_type, exc_obj, tb = sys.exc_info()
+            line_number = tb.tb_lineno
+
+            _LOGGER.error(
+                f"Failed to validate unit settings, Error: {ex}, Line: {line_number}"
+            )
+
     @staticmethod
     def _get_last_reset(uptime):
         now = datetime.now().timestamp()