Add SFR Box integration (#84780)

* Add SFR Box integration

* Adjust error handling in config flow

* Add tests

* Use value_fn

* Add translation

* Enable mypy strict typing

* Add ConfigEntryNotReady

* Rename exception

* Fix requirements

.coveragerc

@@ -1118,6 +1118,9 @@ omit =
     homeassistant/components/sesame/lock.py
     homeassistant/components/seven_segments/image_processing.py
     homeassistant/components/seventeentrack/sensor.py
+    homeassistant/components/sfr_box/__init__.py
+    homeassistant/components/sfr_box/coordinator.py
+    homeassistant/components/sfr_box/sensor.py
     homeassistant/components/shiftr/*
     homeassistant/components/shodan/sensor.py
     homeassistant/components/sia/__init__.py

.strict-typing

@@ -257,6 +257,7 @@ homeassistant.components.sensibo.*
 homeassistant.components.sensirion_ble.*
 homeassistant.components.sensor.*
 homeassistant.components.senz.*
+homeassistant.components.sfr_box.*
 homeassistant.components.shelly.*
 homeassistant.components.simplepush.*
 homeassistant.components.simplisafe.*

CODEOWNERS

@@ -1025,6 +1025,8 @@ build.json @home-assistant/supervisor
 /tests/components/senz/ @milanmeu
 /homeassistant/components/serial/ @fabaff
 /homeassistant/components/seven_segments/ @fabaff
+/homeassistant/components/sfr_box/ @epenet
+/tests/components/sfr_box/ @epenet
 /homeassistant/components/sharkiq/ @JeffResc @funkybunch @AritroSaha10
 /tests/components/sharkiq/ @JeffResc @funkybunch @AritroSaha10
 /homeassistant/components/shell_command/ @home-assistant/core

homeassistant/components/sfr_box/__init__.py

@@ -0,0 +1,53 @@
+"""SFR Box."""
+from __future__ import annotations
+
+from sfrbox_api.bridge import SFRBox
+from sfrbox_api.exceptions import SFRBoxError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.httpx_client import get_async_client
+
+from .const import DOMAIN, PLATFORMS
+from .coordinator import DslDataUpdateCoordinator
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up SFR box as config entry."""
+    box = SFRBox(ip=entry.data[CONF_HOST], client=get_async_client(hass))
+    try:
+        system_info = await box.system_get_info()
+    except SFRBoxError as err:
+        raise ConfigEntryNotReady(
+            f"Unable to connect to {entry.data[CONF_HOST]}"
+        ) from err
+    hass.data.setdefault(DOMAIN, {})
+
+    device_registry = dr.async_get(hass)
+    device_registry.async_get_or_create(
+        config_entry_id=entry.entry_id,
+        identifiers={(DOMAIN, system_info.mac_addr)},
+        name="SFR Box",
+        model=system_info.product_id,
+        sw_version=system_info.version_mainfirmware,
+        configuration_url=f"http://{entry.data[CONF_HOST]}",
+    )
+
+    hass.data[DOMAIN][entry.entry_id] = {
+        "box": box,
+        "dsl_coordinator": DslDataUpdateCoordinator(hass, box),
+    }
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+    return unload_ok

homeassistant/components/sfr_box/config_flow.py

@@ -0,0 +1,49 @@
+"""SFR Box config flow."""
+from __future__ import annotations
+
+from sfrbox_api.bridge import SFRBox
+from sfrbox_api.exceptions import SFRBoxError
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow
+from homeassistant.const import CONF_HOST
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.httpx_client import get_async_client
+
+from .const import DEFAULT_HOST, DOMAIN
+
+DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST, default=DEFAULT_HOST): str,
+    }
+)
+
+
+class SFRBoxFlowHandler(ConfigFlow, domain=DOMAIN):
+    """SFR Box config flow."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, str] | None = None
+    ) -> FlowResult:
+        """Handle a flow initialized by the user."""
+        errors = {}
+        if user_input is not None:
+            try:
+                box = SFRBox(
+                    ip=user_input[CONF_HOST],
+                    client=get_async_client(self.hass),
+                )
+                system_info = await box.system_get_info()
+            except SFRBoxError:
+                errors["base"] = "unknown"
+            else:
+                await self.async_set_unique_id(system_info.mac_addr)
+                self._abort_if_unique_id_configured()
+                self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
+                return self.async_create_entry(title="SFR Box", data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=DATA_SCHEMA, errors=errors
+        )

homeassistant/components/sfr_box/const.py

@@ -0,0 +1,8 @@
+"""SFR Box constants."""
+from homeassistant.const import Platform
+
+DEFAULT_HOST = "192.168.0.1"
+
+DOMAIN = "sfr_box"
+
+PLATFORMS = [Platform.SENSOR]

homeassistant/components/sfr_box/coordinator.py

@@ -0,0 +1,25 @@
+"""SFR Box coordinator."""
+from datetime import timedelta
+import logging
+
+from sfrbox_api.bridge import SFRBox
+from sfrbox_api.models import DslInfo
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+_LOGGER = logging.getLogger(__name__)
+_SCAN_INTERVAL = timedelta(minutes=1)
+
+
+class DslDataUpdateCoordinator(DataUpdateCoordinator[DslInfo]):
+    """Coordinator to manage data updates."""
+
+    def __init__(self, hass: HomeAssistant, box: SFRBox) -> None:
+        """Initialize coordinator."""
+        self._box = box
+        super().__init__(hass, _LOGGER, name="dsl", update_interval=_SCAN_INTERVAL)
+
+    async def _async_update_data(self) -> DslInfo:
+        """Update data."""
+        return await self._box.dsl_get_info()

homeassistant/components/sfr_box/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "sfr_box",
+  "name": "SFR Box",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/sfr_box",
+  "requirements": ["sfrbox-api==0.0.1"],
+  "codeowners": ["@epenet"],
+  "iot_class": "local_polling",
+  "integration_type": "device"
+}

homeassistant/components/sfr_box/sensor.py

@@ -0,0 +1,182 @@
+"""SFR Box sensor platform."""
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from sfrbox_api.bridge import SFRBox
+from sfrbox_api.models import DslInfo, SystemInfo
+
+from homeassistant.components.sensor import (
+    SensorDeviceClass,
+    SensorEntity,
+    SensorEntityDescription,
+    SensorStateClass,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import SIGNAL_STRENGTH_DECIBELS, UnitOfDataRate
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import StateType
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import DslDataUpdateCoordinator
+
+
+@dataclass
+class SFRBoxSensorMixin:
+    """Mixin for SFR Box sensors."""
+
+    value_fn: Callable[[DslInfo], StateType]
+
+
+@dataclass
+class SFRBoxSensorEntityDescription(SensorEntityDescription, SFRBoxSensorMixin):
+    """Description for SFR Box sensors."""
+
+
+SENSOR_TYPES: tuple[SFRBoxSensorEntityDescription, ...] = (
+    SFRBoxSensorEntityDescription(
+        key="linemode",
+        name="Line mode",
+        has_entity_name=True,
+        value_fn=lambda x: x.linemode,
+    ),
+    SFRBoxSensorEntityDescription(
+        key="counter",
+        name="Counter",
+        has_entity_name=True,
+        value_fn=lambda x: x.counter,
+    ),
+    SFRBoxSensorEntityDescription(
+        key="crc",
+        name="CRC",
+        has_entity_name=True,
+        value_fn=lambda x: x.crc,
+    ),
+    SFRBoxSensorEntityDescription(
+        key="noise_down",
+        name="Noise down",
+        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
+        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS,
+        state_class=SensorStateClass.MEASUREMENT,
+        has_entity_name=True,
+        value_fn=lambda x: x.noise_down,
+    ),
+    SFRBoxSensorEntityDescription(
+        key="noise_up",
+        name="Noise up",
+        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
+        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS,
+        state_class=SensorStateClass.MEASUREMENT,
+        has_entity_name=True,
+        value_fn=lambda x: x.noise_up,
+    ),
+    SFRBoxSensorEntityDescription(
+        key="attenuation_down",
+        name="Attenuation down",
+        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
+        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS,
+        state_class=SensorStateClass.MEASUREMENT,
+        has_entity_name=True,
+        value_fn=lambda x: x.attenuation_down,
+    ),
+    SFRBoxSensorEntityDescription(
+        key="attenuation_up",
+        name="Attenuation up",
+        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
+        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS,
+        state_class=SensorStateClass.MEASUREMENT,
+        has_entity_name=True,
+        value_fn=lambda x: x.attenuation_up,
+    ),
+    SFRBoxSensorEntityDescription(
+        key="rate_down",
+        name="Rate down",
+        device_class=SensorDeviceClass.DATA_RATE,
+        native_unit_of_measurement=UnitOfDataRate.KILOBITS_PER_SECOND,
+        state_class=SensorStateClass.MEASUREMENT,
+        has_entity_name=True,
+        value_fn=lambda x: x.rate_down,
+    ),
+    SFRBoxSensorEntityDescription(
+        key="rate_up",
+        name="Rate up",
+        device_class=SensorDeviceClass.DATA_RATE,
+        native_unit_of_measurement=UnitOfDataRate.KILOBITS_PER_SECOND,
+        state_class=SensorStateClass.MEASUREMENT,
+        has_entity_name=True,
+        value_fn=lambda x: x.rate_up,
+    ),
+    SFRBoxSensorEntityDescription(
+        key="line_status",
+        name="Line status",
+        device_class=SensorDeviceClass.ENUM,
+        options=[
+            "No Defect",
+            "Of Frame",
+            "Loss Of Signal",
+            "Loss Of Power",
+            "Loss Of Signal Quality",
+            "Unknown",
+        ],
+        has_entity_name=True,
+        value_fn=lambda x: x.line_status,
+    ),
+    SFRBoxSensorEntityDescription(
+        key="training",
+        name="Training",
+        device_class=SensorDeviceClass.ENUM,
+        options=[
+            "Idle",
+            "G.994 Training",
+            "G.992 Started",
+            "G.922 Channel Analysis",
+            "G.992 Message Exchange",
+            "G.993 Started",
+            "G.993 Channel Analysis",
+            "G.993 Message Exchange",
+            "Showtime",
+            "Unknown",
+        ],
+        has_entity_name=True,
+        value_fn=lambda x: x.training,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up the sensors."""
+    data = hass.data[DOMAIN][entry.entry_id]
+    box: SFRBox = data["box"]
+    system_info = await box.system_get_info()
+
+    entities = [
+        SFRBoxSensor(data["dsl_coordinator"], description, system_info)
+        for description in SENSOR_TYPES
+    ]
+    async_add_entities(entities, True)
+
+
+class SFRBoxSensor(CoordinatorEntity[DslDataUpdateCoordinator], SensorEntity):
+    """SFR Box sensor."""
+
+    entity_description: SFRBoxSensorEntityDescription
+
+    def __init__(
+        self,
+        coordinator: DslDataUpdateCoordinator,
+        description: SFRBoxSensorEntityDescription,
+        system_info: SystemInfo,
+    ) -> None:
+        """Initialize the sensor."""
+        super().__init__(coordinator)
+        self.entity_description = description
+        self._attr_unique_id = f"{system_info.mac_addr}_dsl_{description.key}"
+        self._attr_device_info = {"identifiers": {(DOMAIN, system_info.mac_addr)}}
+
+    @property
+    def native_value(self) -> StateType:
+        """Return the native value of the device."""
+        return self.entity_description.value_fn(self.coordinator.data)

homeassistant/components/sfr_box/strings.json

@@ -0,0 +1,17 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "host": "[%key:common::config_flow::data::host%]"
+        }
+      }
+    },
+    "error": {
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    }
+  }
+}

homeassistant/components/sfr_box/translations/en.json

@@ -0,0 +1,17 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "host": "Host"
+        }
+      }
+    },
+    "error": {
+      "unknown": "Unexpected error"
+    },
+    "abort": {
+      "already_configured": "Device is already configured"
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -363,6 +363,7 @@ FLOWS = {
         "sensorpush",
         "sentry",
         "senz",
+        "sfr_box",
         "sharkiq",
         "shelly",
         "shopping_list",

homeassistant/generated/integrations.json

@@ -4740,6 +4740,12 @@
       "config_flow": false,
       "iot_class": "cloud_polling"
     },
+    "sfr_box": {
+      "name": "SFR Box",
+      "integration_type": "device",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "sharkiq": {
       "name": "Shark IQ",
       "integration_type": "hub",

mypy.ini

@@ -2324,6 +2324,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.sfr_box.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.shelly.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

requirements_all.txt

@@ -2283,6 +2283,9 @@ sensorpush-ble==1.5.2
 # homeassistant.components.sentry
 sentry-sdk==1.12.1
 
+# homeassistant.components.sfr_box
+sfrbox-api==0.0.1
+
 # homeassistant.components.sharkiq
 sharkiq==0.0.1
 

requirements_test_all.txt

@@ -1592,6 +1592,9 @@ sensorpush-ble==1.5.2
 # homeassistant.components.sentry
 sentry-sdk==1.12.1
 
+# homeassistant.components.sfr_box
+sfrbox-api==0.0.1
+
 # homeassistant.components.sharkiq
 sharkiq==0.0.1
 

tests/components/sfr_box/__init__.py

@@ -0,0 +1 @@
+"""Tests for the SFR Box integration."""

tests/components/sfr_box/conftest.py

@@ -0,0 +1,24 @@
+"""Provide common SFR Box fixtures."""
+import pytest
+
+from homeassistant.components.sfr_box.const import DOMAIN
+from homeassistant.config_entries import SOURCE_USER, ConfigEntry
+from homeassistant.const import CONF_HOST
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+@pytest.fixture(name="config_entry")
+def get_config_entry(hass: HomeAssistant) -> ConfigEntry:
+    """Create and register mock config entry."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        source=SOURCE_USER,
+        data={CONF_HOST: "192.168.0.1"},
+        unique_id="e4:5d:51:00:11:22",
+        options={},
+        entry_id="123456",
+    )
+    config_entry.add_to_hass(hass)
+    return config_entry

tests/components/sfr_box/fixtures/system_getInfo.json

@@ -0,0 +1,17 @@
+{
+  "product_id": "NB6VAC-FXC-r0",
+  "mac_addr": "e4:5d:51:00:11:22",
+  "net_mode": "router",
+  "net_infra": "adsl",
+  "uptime": 2353575,
+  "version_mainfirmware": "NB6VAC-MAIN-R4.0.44k",
+  "version_rescuefirmware": "NB6VAC-MAIN-R4.0.44k",
+  "version_bootloader": "NB6VAC-BOOTLOADER-R4.0.8",
+  "version_dsldriver": "NB6VAC-XDSL-A2pv6F039p",
+  "current_datetime": "202212282233",
+  "refclient": "",
+  "idur": "RP3P85K",
+  "alimvoltage": 12251,
+  "temperature": 27560,
+  "serial_number": "XU1001001001001001"
+}

tests/components/sfr_box/test_config_flow.py

@@ -0,0 +1,130 @@
+"""Test the SFR Box config flow."""
+import json
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sfrbox_api.exceptions import SFRBoxError
+from sfrbox_api.models import SystemInfo
+
+from homeassistant import config_entries, data_entry_flow
+from homeassistant.components.sfr_box.const import DOMAIN
+from homeassistant.const import CONF_HOST
+from homeassistant.core import HomeAssistant
+
+from tests.common import load_fixture
+
+
+@pytest.fixture(autouse=True, name="mock_setup_entry")
+def override_async_setup_entry() -> AsyncMock:
+    """Override async_setup_entry."""
+    with patch(
+        "homeassistant.components.sfr_box.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        yield mock_setup_entry
+
+
+async def test_config_flow(hass: HomeAssistant, mock_setup_entry: AsyncMock):
+    """Test we get the form."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == data_entry_flow.FlowResultType.FORM
+    assert result["errors"] == {}
+
+    with patch(
+        "homeassistant.components.sfr_box.config_flow.SFRBox.system_get_info",
+        side_effect=SFRBoxError,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                CONF_HOST: "192.168.0.1",
+            },
+        )
+
+    assert result["type"] == data_entry_flow.FlowResultType.FORM
+    assert result["errors"] == {"base": "unknown"}
+
+    system_info = SystemInfo(**json.loads(load_fixture("system_getInfo.json", DOMAIN)))
+    with patch(
+        "homeassistant.components.sfr_box.config_flow.SFRBox.system_get_info",
+        return_value=system_info,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                CONF_HOST: "192.168.0.1",
+            },
+        )
+
+    assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
+    assert result["title"] == "SFR Box"
+    assert result["data"][CONF_HOST] == "192.168.0.1"
+
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+@pytest.mark.usefixtures("config_entry")
+async def test_config_flow_duplicate_host(
+    hass: HomeAssistant, mock_setup_entry: AsyncMock
+):
+    """Test abort if unique_id configured."""
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 1
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == data_entry_flow.FlowResultType.FORM
+    assert result["errors"] == {}
+
+    system_info = SystemInfo(**json.loads(load_fixture("system_getInfo.json", DOMAIN)))
+    # Ensure mac doesn't match existing mock entry
+    system_info.mac_addr = "aa:bb:cc:dd:ee:ff"
+    with patch(
+        "homeassistant.components.sfr_box.config_flow.SFRBox.system_get_info",
+        return_value=system_info,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                CONF_HOST: "192.168.0.1",
+            },
+        )
+
+    assert result["type"] == data_entry_flow.FlowResultType.ABORT
+    assert result["reason"] == "already_configured"
+    await hass.async_block_till_done()
+
+    assert len(mock_setup_entry.mock_calls) == 0
+
+
+@pytest.mark.usefixtures("config_entry")
+async def test_config_flow_duplicate_mac(
+    hass: HomeAssistant, mock_setup_entry: AsyncMock
+):
+    """Test abort if unique_id configured."""
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 1
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == data_entry_flow.FlowResultType.FORM
+    assert result["errors"] == {}
+
+    system_info = SystemInfo(**json.loads(load_fixture("system_getInfo.json", DOMAIN)))
+    with patch(
+        "homeassistant.components.sfr_box.config_flow.SFRBox.system_get_info",
+        return_value=system_info,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                CONF_HOST: "192.168.0.2",
+            },
+        )
+
+    assert result["type"] == data_entry_flow.FlowResultType.ABORT
+    assert result["reason"] == "already_configured"
+    await hass.async_block_till_done()
+
+    assert len(mock_setup_entry.mock_calls) == 0