feat: port profile switches and networks cache; enable reactivation with native_networkconf_id (refs #99, #87)

Author: sirkirby

custom_components/unifi_network_rules/coordinator.py

@@ -8,10 +8,9 @@
 
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
-from homeassistant.helpers.dispatcher import async_dispatcher_connect, async_dispatcher_send
+from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.entity_registry import async_get as async_get_entity_registry
-from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import Platform
 
 from aiounifi.models.traffic_route import TrafficRoute
@@ -22,14 +21,16 @@
 from aiounifi.models.wlan import Wlan
 from aiounifi.models.device import Device
 
-from .const import DOMAIN, LOGGER, CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL, DEBUG_WEBSOCKET, LOG_TRIGGERS
+from .const import DOMAIN, LOGGER, DEFAULT_UPDATE_INTERVAL, LOG_TRIGGERS
 from .udm import UDMAPI
-from .websocket import SIGNAL_WEBSOCKET_MESSAGE, UnifiRuleWebsocket
-from .helpers.rule import get_rule_id, get_rule_name, get_rule_enabled, get_child_unique_id
+from .websocket import UnifiRuleWebsocket
+from .helpers.rule import get_rule_id, get_child_unique_id
 from .utils.logger import log_data, log_websocket
 from .models.firewall_rule import FirewallRule
 from .models.qos_rule import QoSRule
 from .models.vpn_config import VPNConfig
+from .models.port_profile import PortProfile
+from .models.network import NetworkConf
 
 # This is a fallback if no update_interval is specified
 SCAN_INTERVAL = timedelta(seconds=60)
@@ -110,6 +111,8 @@ def __init__(
         self.vpn_clients: List[VPNConfig] = []
         self.vpn_servers: List[VPNConfig] = []
         self.devices: List[Device] = []  # For LED toggle switches
+        self.port_profiles: List[PortProfile] = []
+        self.networks: List[NetworkConf] = []
 
         # For dynamic entity creation
         self.async_add_entities_callback: AddEntitiesCallback | None = None
@@ -370,6 +373,8 @@ async def _async_update_data(self) -> Dict[str, List[Any]]:
                     "vpn_clients": [],
                     "vpn_servers": [],
                     "devices": [],
+                    "port_profiles": [],
+                    "networks": [],
                 }
 
                 # Store the previous data to detect deletions and protect against API failures
@@ -418,7 +423,7 @@ async def _async_update_data(self) -> Dict[str, List[Any]]:
                             LOGGER.warning("Authentication failure detected during initial fetch: %s", error_msg)
                             # Trigger auth recovery but continue trying other endpoints
                             if hasattr(self.api, "handle_auth_failure"):
-                                recovery_task = asyncio.create_task(self.api.handle_auth_failure(error_msg))
+                                asyncio.create_task(self.api.handle_auth_failure(error_msg))
 
                             # Preserve previous port forwards data if available
                             if previous_data and "port_forwards" in previous_data and previous_data["port_forwards"]:
@@ -467,6 +472,14 @@ async def _async_update_data(self) -> Dict[str, List[Any]]:
 
                 # Then devices (for LED switches)
                 await self._update_devices_in_dict(rules_data)
+                await asyncio.sleep(api_call_delay)
+
+                # Then port profiles
+                await self._update_port_profiles_in_dict(rules_data)
+                await asyncio.sleep(api_call_delay)
+
+                # Then networks
+                await self._update_networks_in_dict(rules_data)
 
                 # Verify the data is valid - check if we have at least some data in key categories
                 # This helps prevent entity removal during temporary API errors
@@ -582,6 +595,8 @@ async def _async_update_data(self) -> Dict[str, List[Any]]:
                     self.vpn_clients = rules_data.get("vpn_clients", [])
                     self.vpn_servers = rules_data.get("vpn_servers", [])
                     self.devices = rules_data.get("devices", [])
+                    self.port_profiles = rules_data.get("port_profiles", [])
+                    self.networks = rules_data.get("networks", [])
 
                     LOGGER.info("Rule collections after refresh: Port Forwards=%d, Traffic Routes=%d, Firewall Policies=%d, Traffic Rules=%d, Legacy Firewall Rules=%d, WLANs=%d, QoS Rules=%d, VPN Clients=%d, VPN Servers=%d, Devices=%d", 
                                len(self.port_forwards),
@@ -601,10 +616,11 @@ async def _async_update_data(self) -> Dict[str, List[Any]]:
                 LOGGER.error("Error updating coordinator data: %s", err)
 
                 # Check if this is an authentication error
-                auth_error = False
+                # Track auth errors for control flow only (no use afterwards)
+                _auth_error = False
                 error_str = str(err).lower()
                 if "401 unauthorized" in error_str or "403 forbidden" in error_str:
-                    auth_error = True
+                    _auth_error = True
                     self._auth_failures += 1
                     self._authentication_in_progress = True
                     try:
@@ -659,6 +675,8 @@ def _check_for_deleted_rules(self, new_data: Dict[str, List[Any]]) -> None:
             "wlans",
             "vpn_clients",
             "vpn_servers",
+            "port_profiles",
+            "networks",
         ]
 
         for rule_type in all_rule_sources_types:
@@ -845,6 +863,49 @@ async def _update_devices_in_dict(self, data: Dict[str, List[Any]]) -> None:
             if not hasattr(self, 'devices'):
                 self.devices = []
 
+    async def _update_port_profiles_in_dict(self, data: Dict[str, List[Any]]) -> None:
+        """Update port profiles in the data dictionary and convert to typed objects."""
+        try:
+            LOGGER.info("Fetching port profiles...")
+            future = await self.api.queue_api_operation(self.api.get_port_profiles)
+            profiles = await future if hasattr(future, "__await__") else future
+            typed: List[PortProfile] = []
+            for item in profiles or []:
+                try:
+                    typed.append(PortProfile(item))
+                except Exception as err:
+                    LOGGER.warning("Error converting port profile: %s", err)
+            data["port_profiles"] = typed
+            self.port_profiles = typed
+            LOGGER.info("Updated %d port profiles", len(typed))
+        except Exception as err:
+            LOGGER.error("Failed to update port profiles: %s", err)
+            data["port_profiles"] = []
+            if not hasattr(self, 'port_profiles'):
+                self.port_profiles = []
+
+    async def _update_networks_in_dict(self, data: Dict[str, List[Any]]) -> None:
+        """Update networks list in the data dictionary and convert to typed objects."""
+        try:
+            LOGGER.info("Fetching networks...")
+            future = await self.api.queue_api_operation(self.api.get_networks)
+            networks = await future if hasattr(future, "__await__") else future
+            typed: List[NetworkConf] = []
+            for item in networks or []:
+                try:
+                    # item is already NetworkConf from API layer
+                    typed.append(item)
+                except Exception as err:
+                    LOGGER.warning("Error converting network: %s", err)
+            data["networks"] = typed
+            self.networks = typed
+            LOGGER.info("Updated %d networks", len(typed))
+        except Exception as err:
+            LOGGER.error("Failed to update networks: %s", err)
+            data["networks"] = []
+            if not hasattr(self, 'networks'):
+                self.networks = []
+
     async def _update_rule_type(self, rule_type: str, fetch_method: Callable) -> None:
         """Update a specific rule type in self.data.
         
@@ -994,7 +1055,7 @@ async def _force_refresh_with_cache_clear(self) -> None:
             log_websocket("Starting forced refresh after rule change detected")
 
             # Store previous data for deletion detection
-            previous_data = self.data.copy() if self.data else {}
+            _unused_previous = self.data.copy() if self.data else {}
 
             # Clear the API cache to ensure we get fresh data
             # But do so without disrupting authentication

custom_components/unifi_network_rules/helpers/rule.py

@@ -16,6 +16,7 @@
 from ..models.firewall_rule import FirewallRule
 from ..models.qos_rule import QoSRule
 from ..models.vpn_config import VPNConfig
+from ..models.port_profile import PortProfile
 from ..const import DOMAIN
 
 LOGGER = logging.getLogger(__name__)
@@ -35,6 +36,9 @@ def get_rule_enabled(rule: Any) -> bool:
     # Check different rule types and return appropriate enabled status
     if isinstance(rule, (PortForward, TrafficRoute, FirewallPolicy, TrafficRule, Wlan, QoSRule, VPNConfig)):
         return getattr(rule, "enabled", False)
+    # Port Profile enabled state (computed)
+    if isinstance(rule, PortProfile):
+        return rule.enabled
 
     # Special handling for Device LED state
     if isinstance(rule, Device):
@@ -154,6 +158,14 @@ def get_rule_id(rule: Any) -> str | None:
             LOGGER.warning("Device without mac attribute: %s", rule)
             return None
 
+    # Port Profile unique id
+    if isinstance(rule, PortProfile):
+        if rule.id:
+            return f"unr_port_profile_{rule.id}"
+        else:
+            LOGGER.warning("PortProfile without id attribute: %s", rule)
+            return None
+    
     # Dictionary fallback - this should not happen with properly typed data
     if isinstance(rule, dict):
         _id = rule.get("_id") or rule.get("id")
@@ -186,7 +198,8 @@ def get_rule_prefix(rule_type: str) -> str:
         "legacy_firewall_rules": "Legacy Rule",
         "qos_rules": "QoS",
         "wlans": "WLAN",
-        "devices": "Device"
+        "devices": "Device",
+        "port_profiles": "Port Profile"
     }
 
     return rule_types.get(rule_type, "Rule")
@@ -332,6 +345,13 @@ def extract_descriptive_name(rule: Any, coordinator=None) -> str | None:
             return f"{vpn_type} VPN"
 
         return None
+    
+    elif isinstance(rule, PortProfile):
+        # For port profiles, prefer the name
+        name = rule.name
+        if name:
+            return name
+        return None
 
     elif isinstance(rule, Device):
         # For devices, return the device name for LED switches

custom_components/unifi_network_rules/models/network.py

@@ -0,0 +1,32 @@
+"""Typed model for UniFi network (networkconf) entries."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict
+
+
+@dataclass
+class NetworkConf:
+    raw: Dict[str, Any]
+
+    @property
+    def id(self) -> str:
+        return str(self.raw.get("_id") or self.raw.get("id") or "")
+
+    @property
+    def name(self) -> str:
+        return str(self.raw.get("name") or self.raw.get("attr_hidden_id") or f"Network {self.id}")
+
+    @property
+    def purpose(self) -> str:
+        return str(self.raw.get("purpose") or "")
+
+    @property
+    def enabled(self) -> bool:
+        # Some networkconfs may not have enabled; treat presence of ip_subnet or WAN purpose as enabled
+        if "enabled" in self.raw:
+            return bool(self.raw.get("enabled"))
+        return True
+
+

custom_components/unifi_network_rules/models/port_profile.py

@@ -0,0 +1,42 @@
+"""Typed model for UniFi Ethernet Port Profile used by UNR.
+
+Keeps raw dict from controller but exposes typed accessors for id, name,
+and computed enabled state that we use for switch entities.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, Optional
+
+
+@dataclass
+class PortProfile:
+    """Represents a UniFi Port Profile with helper accessors.
+
+    Enabled is a computed concept for UNR: a profile is considered "enabled"
+    when it has a native network assigned and management VLAN is not blocked.
+    """
+
+    raw: Dict[str, Any]
+
+    @property
+    def id(self) -> str:
+        return str(self.raw.get("_id") or self.raw.get("id") or "")
+
+    @property
+    def name(self) -> str:
+        return str(self.raw.get("name") or self.raw.get("description") or f"Port Profile {self.id}")
+
+    @property
+    def enabled(self) -> bool:
+        native = self.raw.get("native_networkconf_id")
+        tagged_mgmt = self.raw.get("tagged_vlan_mgmt")
+        # Treat as enabled if a native network is configured and mgmt VLANs are not blocked
+        return bool(native) and tagged_mgmt not in {"block_all", "block-custom"}
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Return a shallow copy of the raw dict suitable for updates."""
+        return dict(self.raw)
+
+