Merge pull request #40 from brenard/options-flow

Ajout de la possibilité de reconfigurer l'intégration

custom_components/bbox/__init__.py

@@ -22,6 +22,8 @@
 async def async_setup_entry(hass: HomeAssistant, entry: BBoxConfigEntry) -> bool:
     """Set up Bouygues Bbox from a config entry."""
     coordinator = BboxDataUpdateCoordinator(hass, entry)
+    await coordinator.connect()
+    entry.async_on_unload(entry.add_update_listener(coordinator.update_configuration))
     await coordinator.async_config_entry_first_refresh()
     entry.runtime_data = coordinator
 

custom_components/bbox/config_flow.py

@@ -10,62 +10,138 @@
 import voluptuous as vol
 
 from homeassistant import config_entries
+from homeassistant.core import callback
 from homeassistant.data_entry_flow import FlowResult
 from homeassistant.helpers.aiohttp_client import async_create_clientsession
 
-from .const import BBOX_URL, CONF_HOST, CONF_PASSWORD, CONF_USE_TLS, DOMAIN
+from .const import (
+    CONF_DEFAULTS,
+    CONF_HOST,
+    CONF_PASSWORD,
+    CONF_REFRESH_RATE,
+    CONF_USE_TLS,
+    DEFAULT_TITLE,
+    DOMAIN
+)
 
 _LOGGER = logging.getLogger(__name__)
 
-DATA_SCHEMA = vol.Schema(
-    {
-        vol.Required(CONF_HOST, default=BBOX_URL): str,
-        vol.Required(CONF_PASSWORD): str,
-        vol.Required(CONF_USE_TLS, default=True): bool,
-    }
-)
 
+class BaseConfigFlow:
+
+    async def async_check_user_input(self, user_input: Mapping[str, Any] | None) -> str | None:
+        """
+        Check Bbox connection with user input
+        Return Bbox serial number on success and set self._errors
+        otherwise.
+        """
+        try:
+            api = Bbox(
+                hostname=user_input[CONF_HOST],
+                password=user_input[CONF_PASSWORD],
+                session=async_create_clientsession(self.hass),
+                use_tls=user_input[CONF_USE_TLS],
+            )
+            await api.async_login()
+            infos = await api.device.async_get_bbox_summary()
+
+            try:
+                if len(infos) > 0:
+                    serialnumber = infos[0]["device"]["serialnumber"]
+                    return serialnumber             
+              raise HttpRequestError("Serial number of device not found")
+
+        except HttpRequestError as err:
+            _LOGGER.warning("Can not to connect at Bbox: %s", err)
+            self._errors["base"] = "cannot_connect"
+        except InvalidAuth as err:
+            _LOGGER.warning("Fail to authenticate to the Bbox: %s", err)
+            self._errors["base"] = "invalid_auth"
+        except BboxException as err:
+            _LOGGER.exception("Unknown error connecting to the Bbox: %s", err)
+            self._errors["base"] = "unknown"
+        return False
 
-class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+
+    @staticmethod
+    def _get_config_schema(defaults=None, password_optional=False):
+        """Get configuration schema"""
+        defaults = defaults or CONF_DEFAULTS
+        return vol.Schema(
+            {
+                vol.Required(CONF_HOST, default=defaults.get(CONF_HOST)): str,
+                (
+                    vol.Optional(CONF_PASSWORD) if password_optional
+                    else vol.Required(CONF_PASSWORD)
+                ): str,
+                vol.Required(CONF_USE_TLS, default=defaults.get(CONF_USE_TLS)): bool,
+                vol.Required(CONF_REFRESH_RATE, default=defaults.get(CONF_REFRESH_RATE)): int,
+            }
+        )
+
+
+class ConfigFlow(BaseConfigFlow, config_entries.ConfigFlow, domain=DOMAIN):
     """Handle a config flow for Bouygues Bbox."""
 
     VERSION = 1
 
+    @staticmethod
+    @callback
+    def async_get_options_flow(
+        config_entry: config_entries.ConfigEntry,
+    ) -> config_entries.OptionsFlow:
+        """Create the options flow."""
+        return OptionsFlow(config_entry)   
+
     async def async_step_user(
         self, user_input: Mapping[str, Any] | None = None
     ) -> FlowResult:
         """Handle the initial step."""
-        errors: dict[str, str] = {}
+        self._errors: dict[str, str] = {}
         if user_input:
-            try:
-                api = Bbox(
-                    hostname=user_input[CONF_HOST],
-                    password=user_input[CONF_PASSWORD],
-                    session=async_create_clientsession(self.hass),
-                    use_tls=user_input[CONF_USE_TLS],
-                )
-                await api.async_login()
-                infos = await api.device.async_get_bbox_summary()
-                if (
-                    len(infos) > 0
-                    and (sn := infos[0].get("device", {}).get("serialnumber")) is None
-                ):
-                    raise HttpRequestError("Serial number of device not found") from err
-                await self.async_set_unique_id(sn)
+            serialnumber = await self.async_check_user_input(user_input)
+            if serialnumber and not self._errors:
+                await self.async_set_unique_id(serialnumber)
                 self._abort_if_unique_id_configured()
+                return self.async_create_entry(title=DEFAULT_TITLE, data=user_input)
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=self._get_config_schema(),
+            errors=self._errors
+        )
+
+class OptionsFlow(BaseConfigFlow, config_entries.OptionsFlow):
+    """Handle a options flow for Bouygues Bbox."""
 
-            except HttpRequestError as err:
-                _LOGGER.warning("Can not to connect at Bbox: %s", err)
-                errors["base"] = "cannot_connect"
-            except AuthorizationError as err:
-                _LOGGER.warning("Fail to authenticate to the Bbox: %s", err)
-                errors["base"] = "invalid_auth"
-            except BboxException:
-                _LOGGER.exception("Unknown error connecting to the Bbox")
-                errors["base"] = "unknown"
-            else:
-                return self.async_create_entry(title="Bouygues Bbox", data=user_input)
+    def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
+        """Initialize options flow."""
+        self.config_entry = config_entry
+
+    async def async_step_init(
+        self, user_input: Mapping[str, Any] | None = None
+    ) -> FlowResult:
+        """Manage the options."""
+        self._errors: dict[str, str] = {}
+        if user_input:
+            if not user_input.get(CONF_PASSWORD):
+                _LOGGER.debug("User do not enter password, keep current configured one")
+                user_input[CONF_PASSWORD] = self.config_entry.data[CONF_PASSWORD]
+            serialnumber = await self.async_check_user_input(user_input)
+            if serialnumber and not self._errors:
+                # update config entry
+                self.hass.config_entries.async_update_entry(
+                    self.config_entry,
+                    data=user_input
+                )
+                # Finish
+                return self.async_create_entry(data=None)
 
         return self.async_show_form(
-            step_id="user", data_schema=DATA_SCHEMA, errors=errors
+            step_id="init",
+            data_schema=self._get_config_schema(
+                self.config_entry.data,
+                password_optional=self.config_entry.data.get(CONF_PASSWORD)
+            ),
+            errors=self._errors
         )

custom_components/bbox/const.py

@@ -1,10 +1,19 @@
 """Constants for the Bouygues Bbox integration."""
 
 DOMAIN = "bbox"
-BBOX_URL = "mabbox.bytel.fr"
+BBOX_NAME = "Bbox"
+MANUFACTURER = "Bouygues"
+DEFAULT_TITLE = f"{MANUFACTURER} {BBOX_NAME}"
 CONF_PASSWORD = "password"
 CONF_HOST = "host"
 CONF_USE_TLS = "use_tls"
+CONF_REFRESH_RATE = "refresh_rate"
+
+CONF_DEFAULTS = {
+    CONF_HOST: "mabbox.bytel.fr",
+    CONF_USE_TLS: True,
+    CONF_REFRESH_RATE: 60,
+}
 TO_REDACT = {
     "username",
     "password",
@@ -19,5 +28,3 @@
     "account",
     "key",
 }
-BBOX_NAME = "Bbox"
-MANUFACTURER = "Bouygues"

custom_components/bbox/coordinator.py

@@ -12,10 +12,9 @@
 from homeassistant.helpers.aiohttp_client import async_create_clientsession
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
-from .const import CONF_HOST, CONF_PASSWORD, CONF_USE_TLS, DOMAIN
+from .const import CONF_DEFAULTS, CONF_HOST, CONF_PASSWORD, CONF_USE_TLS, DOMAIN, CONF_REFRESH_RATE
 
 _LOGGER = logging.getLogger(__name__)
-SCAN_INTERVAL = 60
 
 
 class BboxDataUpdateCoordinator(DataUpdateCoordinator):
@@ -28,14 +27,34 @@ def __init__(
     ) -> None:
         """Class to manage fetching data API."""
         super().__init__(
-            hass, _LOGGER, name=DOMAIN, update_interval=timedelta(seconds=SCAN_INTERVAL)
+            hass, _LOGGER, name=DOMAIN,
+            update_interval=timedelta(
+                seconds=entry.data.get(CONF_REFRESH_RATE, CONF_DEFAULTS[CONF_REFRESH_RATE])
+            )
         )
+        self.entry = entry
+
+    async def connect(self):
+        """Start Bbox connection"""
         self.bbox = Bbox(
-            password=entry.data[CONF_PASSWORD],
-            hostname=entry.data[CONF_HOST],
+            password=self.entry.data[CONF_PASSWORD],
+            hostname=self.entry.data[CONF_HOST],
             session=async_create_clientsession(self.hass),
-            use_tls=entry.data.get(CONF_USE_TLS, False),
+            use_tls=self.entry.data.get(CONF_USE_TLS, False),
+        )
+
+    async def update_configuration(self, hass, entry):
+        """Update configuration"""
+        self.entry = entry
+        await self.connect()
+
+        self.update_interval = timedelta(
+            seconds=entry.data[CONF_REFRESH_RATE]
         )
+        _LOGGER.debug("Coordinator refresh interval updated (%s)", self.update_interval)
+
+        _LOGGER.debug("Force update")
+        await self.async_refresh()
 
     async def _async_update_data(self) -> dict[str, dict[str, Any]]:
         """Fetch data."""

custom_components/bbox/translations/en.json

@@ -5,16 +5,37 @@
         "title": "Bouygue Bbox registration",
         "data": {
           "host": "Host",
-          "password": "Password"
+          "password": "Password",
+          "use_tls": "Use TLS",
+          "refresh_rate": "Refresh rate (in seconds)"
         }
       }
     },
     "error": {
       "cannot_connect": "Failed to connect",
-      "invalid_auth": "Invalid authentication"
+      "invalid_auth": "Invalid authentication",
+      "serialnumber": "Failed to retrieve Bbox serialnumber via its API"
     },
     "abort": {
       "already_configured": "Your account is already configured."
     }
+  },
+  "options": {
+    "step": {
+      "init": {
+        "description": "Please reconfigure the connection with your Bbox.",
+        "data": {
+          "host": "Host",
+          "password": "Password",
+          "use_tls": "Use TLS",
+          "refresh_rate": "Refresh rate (in seconds)"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "Failed to connect",
+      "invalid_auth": "Invalid authentication",
+      "serialnumber": "Failed to retrieve Bbox serialnumber via its API"
+    }
   }
 }

custom_components/bbox/translations/fr.json

@@ -4,17 +4,38 @@
       "user": {
         "title": "Enregistrement Bouygue Bbox",
         "data": {
-          "host": "Url",
-          "password": "Mot de passe"
+          "host": "URL",
+          "password": "Mot de passe",
+          "use_tls": "Utiliser TLS",
+          "refresh_rate": "Fréquence de rafraîchissement (en secondes)"
         }
       }
     },
     "error": {
       "cannot_connect": "Impossible de se connecter",
-      "invalid_auth": "Authentification incorrect"
+      "invalid_auth": "Authentification incorrect",
+      "serialnumber": "Impossible de récupérer le numéro de série de la Bbox via son API"
     },
     "abort": {
       "already_configured": "Votre équipement est déjà configuré."
     }
+  },
+  "options": {
+    "step": {
+      "init": {
+        "description": "Merci de reconfigurer la connexion à votre  Bbox.",
+        "data": {
+          "host": "URL",
+          "password": "Mot de passe",
+          "use_tls": "Utiliser TLS",
+          "refresh_rate": "Fréquence de rafraîchissement (en secondes)"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "Failed to connect",
+      "invalid_auth": "Invalid authentication",
+      "serialnumber": "Failed to retrieve Bbox serialnumber via its API"
+    }
   }
 }