Add Fing integration

CODEOWNERS

@@ -463,6 +463,8 @@ build.json @home-assistant/supervisor
 /tests/components/filesize/ @gjohansson-ST
 /homeassistant/components/filter/ @dgomes
 /tests/components/filter/ @dgomes
+/homeassistant/components/fing/ @Lorenzo-Gasparini
+/tests/components/fing/ @Lorenzo-Gasparini
 /homeassistant/components/fireservicerota/ @cyberjunky
 /tests/components/fireservicerota/ @cyberjunky
 /homeassistant/components/firmata/ @DaAwesomeP

homeassistant/components/fing/__init__.py

@@ -0,0 +1,42 @@
+"""The Fing integration."""
+
+from __future__ import annotations
+
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .coordinator import FingDataUpdateCoordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORMS = [Platform.DEVICE_TRACKER]
+
+type FingConfigEntry = ConfigEntry[FingDataUpdateCoordinator]
+
+
+async def async_setup_entry(hass: HomeAssistant, config_entry: FingConfigEntry) -> bool:
+    """Set up the Fing component."""
+
+    coordinator = FingDataUpdateCoordinator(hass, config_entry)
+    await coordinator.async_config_entry_first_refresh()
+
+    if coordinator.data.get_network_id() is None:
+        _LOGGER.warning(
+            "Skip setting up Fing integration; Received an empty NetworkId from the request - Check if the API version is the latest"
+        )
+        return False
+
+    config_entry.runtime_data = coordinator
+    await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, config_entry: FingConfigEntry
+) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(config_entry, PLATFORMS)

homeassistant/components/fing/config_flow.py

@@ -0,0 +1,133 @@
+"""Config flow file."""
+
+import logging
+from typing import Any
+
+from fing_agent_api import FingAgent
+import httpx
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_NAME
+from homeassistant.core import HomeAssistant
+
+from .const import AGENT_IP, AGENT_KEY, AGENT_PORT, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def _get_data_schema(
+    hass: HomeAssistant, user_input: dict[str, Any] | None = None
+) -> vol.Schema:
+    """Get a schema with default values."""
+
+    if user_input is None:
+        return vol.Schema(
+            {
+                vol.Required(CONF_NAME, default="Fing Agent"): str,
+                vol.Required(AGENT_IP): str,
+                vol.Required(AGENT_PORT, default="49090"): str,
+                vol.Required(AGENT_KEY): str,
+            }
+        )
+
+    return vol.Schema(
+        {
+            vol.Required(CONF_NAME, default=user_input.get(CONF_NAME)): str,
+            vol.Required(AGENT_IP, default=user_input.get(AGENT_IP)): str,
+            vol.Required(AGENT_PORT, default=user_input.get(AGENT_PORT)): str,
+            vol.Required(AGENT_KEY, default=user_input.get(AGENT_KEY)): str,
+        }
+    )
+
+
+class FingConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Fing config flow."""
+
+    # The schema version of the entries that it creates
+    # Home Assistant will call your migrate method if the version changes
+    VERSION = 1
+
+    def exception_to_message(
+        self,
+        exception: BaseException | None,
+        errors: dict[str, str],
+        description_placeholders: dict[str, str],
+    ):
+        """Generate error message from the exception."""
+        if exception is None:
+            _LOGGER.error("Unexpected error during ConfigFlow")
+            errors["base"] = "unexpected_error"
+            return
+
+        _LOGGER.error("Exception raised during ConfigFlow", exc_info=exception)
+        if isinstance(exception, httpx.NetworkError):
+            errors["base"] = "cannot_connect"
+            return
+
+        if isinstance(exception, httpx.TimeoutException):
+            errors["base"] = "timeout_connect"
+            return
+
+        if isinstance(exception, httpx.HTTPStatusError):
+            description_placeholders["message"] = (
+                f"{exception.response.status_code} - {exception.response.reason_phrase}"
+            )
+            if exception.response.status_code == 401:
+                errors["base"] = "invalid_api_key"
+                return
+            errors["base"] = "http_status_error"
+            return
+
+        if isinstance(exception, httpx.InvalidURL):
+            errors["base"] = "url_error"
+            return
+
+        errors["base"] = "unexpected_error"
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Set up user step."""
+        errors: dict[str, str] = {}
+        description_placeholders: dict[str, str] = {}
+
+        if user_input is not None and await self._verify_data(
+            user_input, errors, description_placeholders
+        ):
+            return self.async_create_entry(title=user_input[CONF_NAME], data=user_input)
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=_get_data_schema(self.hass, user_input),
+            errors=errors,
+            description_placeholders=description_placeholders,
+        )
+
+    async def _verify_data(
+        self,
+        user_input: dict[str, Any],
+        errors: dict[str, str],
+        description_placeholders: dict[str, str],
+    ) -> bool:
+        """Verify the user data."""
+
+        try:
+            fing_api = FingAgent(
+                user_input[AGENT_IP], user_input[AGENT_PORT], user_input[AGENT_KEY]
+            )
+            response = await fing_api.get_devices()
+            if response.network_id is not None:
+                return True
+
+            errors["base"] = "api_version_error"
+        except (
+            httpx.HTTPError,
+            httpx.InvalidURL,
+            httpx.CookieConflict,
+            httpx.StreamError,
+            Exception,
+        ) as exception:
+            self.exception_to_message(exception, errors, description_placeholders)
+
+        return False

homeassistant/components/fing/const.py

@@ -0,0 +1,6 @@
+"""Const for the Fing integration."""
+
+AGENT_IP = "agent_ip"
+AGENT_PORT = "agent_port"
+AGENT_KEY = "agent_key"
+DOMAIN = "fing"

homeassistant/components/fing/coordinator.py

@@ -0,0 +1,65 @@
+"""DataUpdateCoordinator for Fing integration."""
+
+from datetime import timedelta
+import logging
+from typing import Self
+
+from fing_agent_api import FingAgent
+from fing_agent_api.models import Device
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import AGENT_IP, AGENT_KEY, AGENT_PORT, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class FingDataFetcher:
+    """Keep data from the Fing Agent."""
+
+    def __init__(self, hass: HomeAssistant, ip: str, port: int, key: str) -> None:
+        """Initialize Fing entity data."""
+        self._hass = hass
+        self._fing = FingAgent(ip, port, key)
+        self._network_id = None
+        self._devices: dict[str, Device] = {}
+
+    def get_devices(self):
+        """Return all the devices."""
+        return self._devices
+
+    def get_network_id(self) -> str | None:
+        """Return the network id."""
+        return self._network_id
+
+    async def fetch_data(self) -> Self:
+        """Fecth data from Fing."""
+        response = await self._fing.get_devices()
+        self._network_id = response.network_id
+        self._devices = {device.mac: device for device in response.devices}
+        return self
+
+
+class FingDataUpdateCoordinator(DataUpdateCoordinator[FingDataFetcher]):
+    """Class to manage fetching data from Fing Agent."""
+
+    def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
+        """Initialize global Fing updater."""
+        self._fing_fetcher = FingDataFetcher(
+            hass,
+            config_entry.data[AGENT_IP],
+            int(config_entry.data[AGENT_PORT]),
+            config_entry.data[AGENT_KEY],
+        )
+
+        update_interval = timedelta(seconds=30)
+        super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval)
+
+    async def _async_update_data(self) -> FingDataFetcher:
+        """Fetch data from Fing Agent."""
+        try:
+            return await self._fing_fetcher.fetch_data()
+        except Exception as err:
+            raise UpdateFailed(f"Update failed: {err}") from err

homeassistant/components/fing/device_tracker.py

@@ -0,0 +1,128 @@
+"""Platform for Device tracker integration."""
+
+from fing_agent_api.models import Device
+
+from homeassistant.components.device_tracker import ScannerEntity, SourceType
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import FingConfigEntry
+from .coordinator import FingDataUpdateCoordinator
+from .utils import get_icon_from_type
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: FingConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Add sensors for passed config_entry in HA."""
+    coordinator = config_entry.runtime_data
+    tracked_devices: list[FingTrackedDevice] = []
+
+    entity_registry = er.async_get(hass)
+
+    @callback
+    def add_entities() -> None:
+        new_entities = [
+            FingTrackedDevice(coordinator, device)
+            for device in coordinator.data.get_devices().values()
+        ]
+
+        new_ent_unique_ids = {entity.unique_id for entity in new_entities}
+        prev_ent_unique_ids = {entity.unique_id for entity in tracked_devices}
+
+        entities_to_remove = [
+            entity
+            for entity in tracked_devices
+            if entity.unique_id not in new_ent_unique_ids
+        ]
+
+        entities_to_add = [
+            entity
+            for entity in new_entities
+            if entity.unique_id not in prev_ent_unique_ids
+        ]
+
+        # Removes all the entities that are no more tracked by the agent
+        for entity in entities_to_remove:
+            entity_registry.async_remove(entity.entity_id)
+            tracked_devices.remove(entity)
+
+        # Adds all the new entities tracked by the agent
+        async_add_entities(entities_to_add)
+        tracked_devices.extend(entities_to_add)
+
+    add_entities()
+    config_entry.async_on_unload(coordinator.async_add_listener(add_entities))
+
+
+class FingTrackedDevice(CoordinatorEntity[FingDataUpdateCoordinator], ScannerEntity):
+    """Represent a tracked device."""
+
+    _attr_has_entity_name = True
+
+    def __init__(self, coordinator: FingDataUpdateCoordinator, device: Device) -> None:
+        """Set up FingDevice entity."""
+        super().__init__(coordinator)
+        self._mac = device.mac
+        self._device = coordinator.data.get_devices()[device.mac]
+        self._network_id = coordinator.data.get_network_id()
+        self._attr_name = self._device.name
+
+    @property
+    def mac_address(self) -> str:
+        """Return mac_address."""
+        return self._mac
+
+    @property
+    def unique_id(self) -> str | None:
+        """Return unique ID of the entity."""
+        return f"{self._network_id}-{self.mac_address}"
+
+    @property
+    def icon(self) -> str | None:
+        """Return the icon."""
+        return get_icon_from_type(self._device.type)
+
+    @property
+    def ip_address(self) -> str | None:
+        """Return the primary ip address of the device."""
+        return self._device.ip[0] if self._device.ip else None
+
+    @property
+    def extra_state_attributes(self) -> dict[str, str]:
+        """Return the attributes."""
+        attrs: dict[str, str] = {}
+        if self._device.type:
+            attrs["type"] = self._device.type
+        if self._device.make:
+            attrs["make"] = self._device.make
+        if self._device.model:
+            attrs["model"] = self._device.model
+        return attrs
+
+    @property
+    def entity_registry_enabled_default(self) -> bool:
+        """Return if entity is enabled by default."""
+        return True
+
+    @property
+    def is_connected(self) -> bool:
+        """Return true if the device is connected to the network."""
+        return self._device.active
+
+    @property
+    def source_type(self) -> SourceType:
+        """Return source type."""
+        return SourceType.ROUTER
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        updated_device_data = self.coordinator.data.get_devices().get(self._mac)
+        if updated_device_data is not None:
+            self._device = updated_device_data
+            self.async_write_ha_state()

homeassistant/components/fing/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "fing",
+  "name": "Fing",
+  "codeowners": ["@Lorenzo-Gasparini"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/fing_unit",
+  "iot_class": "local_polling",
+  "requirements": ["fing_agent_api==1.0.1"]
+}