feat: enhance network management with VPN client/server derivation and filtering

- Updated the UnifiRuleUpdateCoordinator to derive VPN clients and servers from existing network data, improving efficiency and accuracy.
- Introduced a new helper function to filter out VPN networks, ensuring only suitable networks are exposed as switch entities.
- Added methods for updating and toggling network configurations in the API, enhancing network management capabilities.
- Improved error handling and logging for network operations.

Author: sirkirby

custom_components/unifi_network_rules/coordinator.py

@@ -454,6 +454,10 @@ async def _async_update_data(self) -> Dict[str, List[Any]]:
                 await self._update_traffic_rules_in_dict(rules_data)
                 await asyncio.sleep(api_call_delay)
 
+                # Then networks (source of truth for VPN derivation)
+                await self._update_networks_in_dict(rules_data)
+                await asyncio.sleep(api_call_delay)
+
                 # Then legacy firewall rules
                 await self._update_legacy_firewall_rules_in_dict(rules_data)
                 await asyncio.sleep(api_call_delay)
@@ -462,11 +466,9 @@ async def _async_update_data(self) -> Dict[str, List[Any]]:
                 await self._update_qos_rules_in_dict(rules_data)
                 await asyncio.sleep(api_call_delay)
 
-                # Then VPN clients
+                # Then derive VPN clients and servers from networks
                 await self._update_vpn_clients_in_dict(rules_data)
                 await asyncio.sleep(api_call_delay)
-                
-                # Then VPN servers
                 await self._update_vpn_servers_in_dict(rules_data)
                 await asyncio.sleep(api_call_delay)
 
@@ -476,10 +478,6 @@ async def _async_update_data(self) -> Dict[str, List[Any]]:
 
                 # 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
@@ -824,12 +822,57 @@ async def _update_qos_rules_in_dict(self, data: Dict[str, List[Any]]) -> None:
         await self._update_rule_type_in_dict(data, "qos_rules", self.api.get_qos_rules)
 
     async def _update_vpn_clients_in_dict(self, data: Dict[str, List[Any]]) -> None:
-        """Update VPN clients in the given data dictionary."""
-        await self._update_rule_type_in_dict(data, "vpn_clients", self.api.get_vpn_clients)
+        """Update VPN clients from already-fetched networks."""
+        try:
+            # Derive from networks if available; otherwise, fall back to API
+            networks = data.get("networks") or self.networks
+            if networks:
+                from .models.vpn_config import VPNConfig
+                clients = []
+                for n in networks:
+                    raw = getattr(n, 'raw', {}) if hasattr(n, 'raw') else {}
+                    purpose = raw.get("purpose", "")
+                    vpn_type = raw.get("vpn_type", "")
+                    is_client = purpose == "vpn-client" or vpn_type in ["openvpn-client", "wireguard-client"]
+                    if is_client:
+                        try:
+                            clients.append(VPNConfig(raw))
+                        except Exception as err:
+                            LOGGER.debug("Skipping VPN client conversion error: %s", err)
+                data["vpn_clients"] = clients
+                return
+            # Fallback to API method if networks missing
+            await self._update_rule_type_in_dict(data, "vpn_clients", self.api.get_vpn_clients)
+        except Exception as err:
+            LOGGER.error("Error deriving VPN clients: %s", err)
+            if "vpn_clients" not in data:
+                data["vpn_clients"] = []
 
     async def _update_vpn_servers_in_dict(self, data: Dict[str, List[Any]]) -> None:
-        """Update VPN servers in the given data dictionary."""
-        await self._update_rule_type_in_dict(data, "vpn_servers", self.api.get_vpn_servers)
+        """Update VPN servers from already-fetched networks."""
+        try:
+            networks = data.get("networks") or self.networks
+            if networks:
+                from .models.vpn_config import VPNConfig
+                servers = []
+                for n in networks:
+                    raw = getattr(n, 'raw', {}) if hasattr(n, 'raw') else {}
+                    purpose = raw.get("purpose", "")
+                    vpn_type = raw.get("vpn_type", "")
+                    is_server = purpose == "vpn-server" or vpn_type in ["openvpn-server", "wireguard-server"]
+                    if is_server:
+                        try:
+                            servers.append(VPNConfig(raw))
+                        except Exception as err:
+                            LOGGER.debug("Skipping VPN server conversion error: %s", err)
+                data["vpn_servers"] = servers
+                return
+            # Fallback to API method if networks missing
+            await self._update_rule_type_in_dict(data, "vpn_servers", self.api.get_vpn_servers)
+        except Exception as err:
+            LOGGER.error("Error deriving VPN servers: %s", err)
+            if "vpn_servers" not in data:
+                data["vpn_servers"] = []
 
     async def _update_devices_in_dict(self, data: Dict[str, List[Any]]) -> None:
         """Update devices in the data dictionary."""

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.network import NetworkConf
 from ..models.port_profile import PortProfile
 from ..const import DOMAIN
 
@@ -36,9 +37,6 @@ 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):
@@ -50,6 +48,14 @@ def get_rule_enabled(rule: Any) -> bool:
             return led_state != 'off'  # True if not explicitly turned off
         return True  # Default to enabled if no override info
 
+    # Networks enabled (corporate LAN typically has 'enabled')
+    if isinstance(rule, NetworkConf):
+        return rule.enabled
+    
+    # Port profile enabled state
+    if isinstance(rule, PortProfile):
+        return rule.enabled
+    
     # For dictionaries, try common enabled attributes
     if isinstance(rule, dict):
         return rule.get("enabled", False)
@@ -158,7 +164,15 @@ def get_rule_id(rule: Any) -> str | None:
             LOGGER.warning("Device without mac attribute: %s", rule)
             return None
 
-    # Port Profile unique id
+    # Handle NetworkConf
+    if isinstance(rule, NetworkConf):
+        if rule.id:
+            return f"unr_network_{rule.id}"
+        else:
+            LOGGER.warning("NetworkConf without id attribute: %s", rule)
+            return None
+    
+    # Handle PortProfile
     if isinstance(rule, PortProfile):
         if rule.id:
             return f"unr_port_profile_{rule.id}"
@@ -199,7 +213,10 @@ def get_rule_prefix(rule_type: str) -> str:
         "qos_rules": "QoS",
         "wlans": "WLAN",
         "devices": "Device",
-        "port_profiles": "Port Profile"
+        "port_profiles": "Port Profile",
+        # For networks we return an empty prefix because the descriptive
+        # name will already include the desired label (e.g., WAN1, VLAN 1).
+        "networks": ""
     }
 
     return rule_types.get(rule_type, "Rule")
@@ -345,13 +362,6 @@ 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
@@ -366,6 +376,41 @@ def extract_descriptive_name(rule: Any, coordinator=None) -> str | None:
     elif isinstance(rule, dict):
         # For dictionaries, try common name attributes
         return rule.get("name") or rule.get("description")
+    
+    elif isinstance(rule, NetworkConf):
+        # Build specialized names for networks:
+        # - WAN: "WAN<idx> <name>" when attr_hidden_id starts with WAN or purpose==wan
+        # - LAN/Corporate: "VLAN <vlan_id> <name>" when vlan_enabled and vlan id available
+        # - Special case: name exactly "WAN Magic" becomes "UniFi WAN Magic"
+        raw = getattr(rule, "raw", {}) if hasattr(rule, "raw") else {}
+        name = raw.get("name") or rule.name
+        hidden_id = raw.get("attr_hidden_id", "") or ""
+        purpose = raw.get("purpose", "") or ""
+
+        # Special case first
+        if name == "WAN Magic":
+            return "UniFi WAN Magic"
+
+        # WAN naming
+        if purpose == "wan" or (isinstance(hidden_id, str) and hidden_id.upper().startswith("WAN")):
+            # Extract index from WAN/WAN2/WAN3 ... when present
+            suffix = ""
+            if isinstance(hidden_id, str) and len(hidden_id) > 3 and hidden_id.upper().startswith("WAN"):
+                suffix = hidden_id[3:]  # characters after WAN
+            wan_label = f"WAN{suffix}" if suffix else "WAN"
+            return f"{wan_label} {name}".strip()
+
+        # LAN/VLAN naming
+        vlan_id = raw.get("vlan") or raw.get("vlan_id")
+        if raw.get("vlan_enabled") and vlan_id is not None:
+            return f"VLAN {vlan_id} {name}".strip()
+
+        # Default LAN (no VLAN) naming
+        if (purpose == "corporate" or (isinstance(hidden_id, str) and hidden_id.upper() == "LAN")) and not raw.get("vlan_enabled"):
+            return f"LAN {name}".strip()
+
+        # Default: return name as-is
+        return name
 
     # For other types, try common attributes
     if hasattr(rule, "name"):
@@ -394,6 +439,25 @@ def get_rule_name(rule: Any, coordinator=None) -> str | None:
         rule_type = "qos_rules"
     elif isinstance(rule, Device):
         rule_type = "devices"
+    elif isinstance(rule, NetworkConf):
+        # Decide if this network should be exposed as a switch entity.
+        # Omit VPN networks since we already have VPN switches.
+        raw = getattr(rule, "raw", {}) if hasattr(rule, "raw") else {}
+        purpose = str(raw.get("purpose", "")).lower()
+        vpn_type = str(raw.get("vpn_type", "")).lower()
+        is_vpn = (
+            purpose.startswith("vpn")
+            or purpose in {"remote-user-vpn", "vpn-client", "vpn-server"}
+            or "vpn" in vpn_type
+            or "wireguard" in vpn_type
+            or "openvpn" in vpn_type
+        )
+        if is_vpn:
+            rule_type = None  # signal to caller there is no switch type
+        else:
+            rule_type = "networks"
+    elif isinstance(rule, PortProfile):
+        rule_type = "port_profiles"
     elif isinstance(rule, dict) and "type" in rule:
         rule_type = rule.get("type")
 
@@ -602,4 +666,31 @@ def is_our_entity(entity_entry, domain=DOMAIN) -> bool:
     """
     # Check if entity's platform matches our domain
     # This property cannot be changed by users
-    return entity_entry.platform == domain
+    return entity_entry.platform == domain
+
+
+# --- Network helpers ---
+def is_vpn_network(network: Any) -> bool:
+    """Return True if a network (dict or NetworkConf) represents a VPN entity.
+
+    Detects both purpose values and vpn_type variants (OpenVPN/WireGuard).
+    """
+    raw = getattr(network, "raw", {}) if hasattr(network, "raw") else (network if isinstance(network, dict) else {})
+    purpose = str(raw.get("purpose", "")).lower()
+    vpn_type = str(raw.get("vpn_type", "")).lower()
+    return (
+        purpose.startswith("vpn")
+        or purpose in {"remote-user-vpn", "vpn-client", "vpn-server"}
+        or "vpn" in vpn_type
+        or "wireguard" in vpn_type
+        or "openvpn" in vpn_type
+    )
+
+
+def filter_switchable_networks(networks: list[Any]) -> list[Any]:
+    """Filter out VPN networks; keep networks suitable for switch entities."""
+    try:
+        return [n for n in networks if not is_vpn_network(n)]
+    except Exception:
+        # Fail-safe: if anything goes wrong, return original list
+        return networks

custom_components/unifi_network_rules/switch.py

@@ -30,10 +30,12 @@
     get_child_entity_name, 
     get_child_unique_id, 
     get_child_entity_id,
-    extract_descriptive_name
+    extract_descriptive_name,
+    filter_switchable_networks,
 )
 from .models.vpn_config import VPNConfig
 from .models.port_profile import PortProfile
+from .models.network import NetworkConf
 
 LOGGER = logging.getLogger(__name__)
 
@@ -110,6 +112,8 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry, asyn
         ("vpn_clients", coordinator.vpn_clients, UnifiVPNClientSwitch),
         ("vpn_servers", coordinator.vpn_servers, UnifiVPNServerSwitch),
         ("port_profiles", coordinator.port_profiles, UnifiPortProfileSwitch),
+        # Filter networks to exclude all VPN forms (purpose and vpn_type)
+        ("networks", filter_switchable_networks(coordinator.networks or []), UnifiNetworkSwitch),
     ]
 
     for rule_type, rules, entity_class in all_rule_sources:
@@ -1823,4 +1827,57 @@ def extra_state_attributes(self) -> Dict[str, Any]:
             attrs["tagged_vlan_mgmt"] = raw.get("tagged_vlan_mgmt")
             attrs["op_mode"] = raw.get("op_mode")
             attrs["poe_mode"] = raw.get("poe_mode")
+        return attrs
+
+class UnifiNetworkSwitch(UnifiRuleSwitch):
+    """Switch to enable/disable a UniFi Network (LAN)."""
+
+    def __init__(
+        self,
+        coordinator: UnifiRuleUpdateCoordinator,
+        rule_data: NetworkConf,
+        rule_type: str = "networks",
+        entry_id: str | None = None,
+    ) -> None:
+        super().__init__(coordinator, rule_data, rule_type, entry_id)
+        self._attr_icon = "mdi:lan"
+
+    async def _async_toggle_rule(self, enable: bool) -> None:
+        network = self._get_current_rule()
+        if network is None:
+            raise HomeAssistantError(f"Cannot find network with ID: {self._rule_id}")
+
+        # Optimistic
+        self.mark_pending_operation(enable)
+        self.async_write_ha_state()
+
+        async def handle_operation_complete(f):
+            try:
+                ok = f.result()
+                if not ok:
+                    self.mark_pending_operation(not enable)
+                self.async_write_ha_state()
+            except Exception:
+                self.mark_pending_operation(not enable)
+                self.async_write_ha_state()
+
+        # Queue via API
+        async def toggle_wrapper(n: NetworkConf):
+            # Force desired enabled in payload
+            n.raw["enabled"] = enable
+            return await self.coordinator.api.update_network(n)
+
+        future = await self.coordinator.api.queue_api_operation(toggle_wrapper, network)
+        future.add_done_callback(lambda f: self.hass.async_create_task(handle_operation_complete(f)))
+
+    @property
+    def extra_state_attributes(self) -> Dict[str, Any]:
+        attrs: Dict[str, Any] = {}
+        net = self._get_current_rule()
+        if net and hasattr(net, "raw"):
+            raw = net.raw
+            attrs["purpose"] = raw.get("purpose")
+            attrs["ip_subnet"] = raw.get("ip_subnet")
+            attrs["vlan_enabled"] = raw.get("vlan_enabled")
+            attrs["networkgroup"] = raw.get("networkgroup")
         return attrs