v0.6.10

custom_components/network_scanner/__init__.py

@@ -1,18 +1,26 @@
-from .sensor import NetworkScanner
+from __future__ import annotations
+import logging
+from homeassistant.core import HomeAssistant
+from homeassistant.config_entries import ConfigEntry
+
 from .const import DOMAIN
 
-async def async_setup(hass, config):
-    """Set up the Network Scanner component."""
-    # Store YAML configuration in hass.data
-    hass.data[DOMAIN] = config.get(DOMAIN, {})
-    return True
+_LOGGER = logging.getLogger(__name__)
+PLATFORMS: list[str] = ["sensor"]
 
-async def async_setup_entry(hass, config_entry):
-    """Set up Network Scanner from a config entry."""
-    await hass.config_entries.async_forward_entry_setups(config_entry, ["sensor"])
+async def async_setup(hass: HomeAssistant, config: dict) -> bool:
+    # Preserve YAML for defaults in config flow (optional)
+    hass.data.setdefault(DOMAIN, config.get(DOMAIN, {}) or {})
     return True
 
-async def async_unload_entry(hass, config_entry):
-    """Unload a config entry."""
-    await hass.config_entries.async_forward_entry_unload(config_entry, "sensor")
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    entry.async_on_unload(entry.add_update_listener(_async_update_listener))
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     return True
+
+async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    await hass.config_entries.async_reload(entry.entry_id)
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    return unload_ok

custom_components/network_scanner/config_flow.py

@@ -1,61 +1,139 @@
+from __future__ import annotations
+import json
+import logging
+from ipaddress import ip_network
 import voluptuous as vol
+from typing import Any, Dict
+
 from homeassistant import config_entries
-from .const import DOMAIN
-import logging
+
+try:
+    from homeassistant.helpers.selector import selector as ha_selector
+    def TextSelector():
+        return ha_selector({"text": {"multiline": True}})
+except Exception:
+    def TextSelector():
+        return str
+
+from .const import DOMAIN, DEFAULT_IP_RANGE
 
 _LOGGER = logging.getLogger(__name__)
 
+DEFAULT_NMAP_ARGS = "-sn -PE -PS22,80,443 -PA80,443 -PU53 -T4"
+
+def _normalise_mac_key(mac: str) -> str:
+    return mac.upper() if isinstance(mac, str) else ""
+
 class NetworkScannerConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
-    """Handle a config flow for Network Scanner."""
+    VERSION = 1
 
     async def async_step_user(self, user_input=None):
-        def format_dict_for_printing(d):
-            return {k: str(v) for k, v in d.items()}
+        yaml_defaults = self.hass.data.get(DOMAIN, {}) or {}
+        schema = vol.Schema({
+            vol.Required("ip_range", description={"suggested_value": yaml_defaults.get("ip_range", DEFAULT_IP_RANGE)}): str,
+            vol.Optional("nmap_args", description={"suggested_value": DEFAULT_NMAP_ARGS}): str,
+            vol.Optional("mac_directory_json_text", description={"suggested_value": ""}): TextSelector(),
+            vol.Optional("mac_directory_json_url", description={"suggested_value": ""}): str,
+        })
 
-        """Manage the configurations from the user interface."""
-        errors = {}
+        if user_input is None:
+            return self.async_show_form(step_id="user", data_schema=schema, errors={})
+
+        errors: Dict[str, str] = {}
+        ipr = (user_input.get("ip_range") or "").strip()
+        try:
+            ip_network(ipr, strict=False)
+        except Exception:
+            errors["ip_range"] = "invalid_ip_range"
 
-        # Load data from configuration.yaml
-        yaml_config = self.hass.data.get(DOMAIN, {})
-        _LOGGER.debug("YAML Config: %s", yaml_config)
+        # Light validation of JSON text
+        jtxt = (user_input.get("mac_directory_json_text") or "").strip()
+        if jtxt:
+            try:
+                parsed = json.loads(jtxt)
+                if not isinstance(parsed, dict):
+                    errors["mac_directory_json_text"] = "invalid_json"
+            except Exception:
+                errors["mac_directory_json_text"] = "invalid_json"
 
-        if user_input is not None:
-            return self.async_create_entry(title="Network Scanner", data=user_input)
+        if errors:
+            return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
 
-        data_schema_dict = {
-            vol.Required("ip_range", description={"suggested_value": yaml_config.get("ip_range", "192.168.1.0/24")}): str
+        # Normalise directory now so the sensor can use it directly
+        directory = {}
+        if jtxt:
+            raw = json.loads(jtxt)
+            block = raw.get("data", raw) if isinstance(raw, dict) else {}
+            if isinstance(block, dict):
+                for k, v in block.items():
+                    mk = _normalise_mac_key(k)
+                    if not mk:
+                        continue
+                    if isinstance(v, dict):
+                        directory[mk] = {"name": str(v.get("name", "")), "desc": str(v.get("desc", ""))}
+                    else:
+                        directory[mk] = {"name": str(v), "desc": ""}
+
+        data = {
+            "ip_range": ipr,
+            "mac_directory": directory,
+            "mac_directory_json_url": (user_input.get("mac_directory_json_url") or "").strip(),
         }
+        return self.async_create_entry(title="Network Scanner Extended", data=data)
+
+    # Options flow mirrors the same fields
+    async def async_step_init(self, user_input=None):
+        return await self.async_step_user(user_input)
+
+    async def async_step_options(self, user_input=None):
+        return await self.async_step_user(user_input)
+
+class NetworkScannerOptionsFlow(config_entries.OptionsFlow):
+    def __init__(self, entry: config_entries.ConfigEntry) -> None:
+        self.entry = entry
+
+    async def async_step_init(self, user_input=None):
+        return await self.async_step_user(user_input)
+
+    async def async_step_user(self, user_input=None):
+        data = self.entry.data or {}
+        opts = self.entry.options or {}
+
+        schema = vol.Schema({
+            vol.Required("ip_range", description={"suggested_value": opts.get("ip_range", data.get("ip_range", DEFAULT_IP_RANGE))}): str,
+            vol.Optional("nmap_args", description={"suggested_value": DEFAULT_NMAP_ARGS}): str,
+            vol.Optional("mac_directory_json_text", description={"suggested_value": opts.get("mac_directory_json_text", "")}): TextSelector(),
+            vol.Optional("mac_directory_json_url", description={"suggested_value": opts.get("mac_directory_json_url", data.get("mac_directory_json_url",""))}): str,
+        })
+
+        if user_input is None:
+            return self.async_show_form(step_id="user", data_schema=schema, errors={})
+
+        # Validate similarly
+        errors = {}
+        try:
+            ip_network((user_input.get("ip_range") or "").strip(), strict=False)
+        except Exception:
+            errors["ip_range"] = "invalid_ip_range"
+
+        jtxt = (user_input.get("mac_directory_json_text") or "").strip()
+        if jtxt:
+            try:
+                parsed = json.loads(jtxt)
+                block = parsed.get("data", parsed) if isinstance(parsed, dict) else {}
+                if not isinstance(block, dict):
+                    errors["mac_directory_json_text"] = "invalid_json"
+            except Exception:
+                errors["mac_directory_json_text"] = "invalid_json"
+
+        if errors:
+            return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
+
+        return self.async_create_entry(title="", data={
+            "ip_range": (user_input.get("ip_range") or "").strip(),
+            "mac_directory_json_text": jtxt,
+            "mac_directory_json_url": (user_input.get("mac_directory_json_url") or "").strip(),
+        })
 
-        # Add mac mappings with values from YAML if available
-        for i in range(1, 26):  # Ensure at least 25 entries
-            key = f"mac_mapping_{i}"
-            if key in yaml_config:
-                suggested_value = yaml_config.get(key)
-                _LOGGER.debug("YAML Config key: %s", suggested_value)
-            else:
-                suggested_value = None  # No value in YAML config
-
-            # Add the optional field to the schema, with or without suggested_value
-            data_schema_dict[vol.Optional(key, description={"suggested_value": suggested_value})] = str
-
-        # Continue to add more mappings if available in the YAML config
-        i = 26
-        while True:
-            key = f"mac_mapping_{i}"
-            if key in yaml_config:
-                suggested_value = yaml_config.get(key)
-                _LOGGER.debug("YAML Config key: %s", suggested_value)
-                data_schema_dict[vol.Optional(key, description={"suggested_value": suggested_value})] = str
-                i += 1
-            else:
-                break  # Exit loop when no more mappings are found in the YAML config
-
-        _LOGGER.debug("schema: %s", format_dict_for_printing(data_schema_dict))
-        data_schema = vol.Schema(data_schema_dict)
-
-        return self.async_show_form(
-            step_id="user",
-            data_schema=data_schema,
-            errors=errors,
-            description_placeholders={"description": "Enter the IP range and MAC mappings"}
-        )
+async def async_get_options_flow(config_entry: config_entries.ConfigEntry):
+    return NetworkScannerOptionsFlow(config_entry)

custom_components/network_scanner/const.py

@@ -1,2 +1,10 @@
-"""Constants for the Network Scanner integration."""
 DOMAIN = "network_scanner"
+
+# Default UI values
+DEFAULT_IP_RANGE = "192.168.1.0/24"
+
+# Card-friendly status strings
+STATUS_IDLE = "idle"
+STATUS_SCANNING = "scanning"
+STATUS_OK = "ok"
+STATUS_ERROR = "error"

custom_components/network_scanner/en.json

@@ -0,0 +1,15 @@
+{
+  "title": "Network Scanner Extended",
+  "config": {
+    "step": {
+      "user": {
+        "title": "Network Scanner Extended",
+        "description": "Enter the IP range and an optional MAC directory (JSON text or URL)."
+      }
+    },
+    "error": {
+      "invalid_json": "That JSON isn’t valid. Paste a JSON object (optionally with a top-level \"data\" object).",
+      "invalid_ip_range": "Please enter a valid CIDR range, e.g. 192.168.1.0/24."
+    }
+  }
+}

custom_components/network_scanner/manifest.json

@@ -1,12 +1,13 @@
 {
   "domain": "network_scanner",
-  "name": "Network Scanner",
-  "codeowners": ["@parvez"],
+  "name": "Network Scanner Extended",
+  "codeowners": ["@paganl"],
   "config_flow": true,
   "dependencies": [],
-  "documentation": "https://github.com/parvez/network_scanner",
-  "iot_class": "local_polling",
-  "issue_tracker": "https://github.com/parvez/network_scanner/issues",
-  "requirements": ["python-nmap"],
-  "version": "1.0.7"
+  "documentation": "https://github.com/paganl/network_scanner_extended",
+  "issue_tracker": "https://github.com/paganl/network_scanner_extended/issues",
+  "requirements": ["python-nmap==0.7.1"],
+  "version": "0.6.9",
+  "integration_type": "hub",
+  "iot_class": "local_polling"
 }