Compatibilité Bbox OS 23.8.16

custom_components/bbox/config_flow.py

@@ -50,8 +50,7 @@ async def async_step_user(
                     len(infos) > 0
                     and (sn := infos[0].get("device", {}).get("serialnumber")) is None
                 ):
-                    raise HttpRequestError("Serial number of device not found")
-
+                    raise HttpRequestError("Serial number of device not found") from err
                 await self.async_set_unique_id(sn)
                 self._abort_if_unique_id_configured()
 

custom_components/bbox/coordinator.py

@@ -41,7 +41,8 @@ async def _async_update_data(self) -> dict[str, dict[str, Any]]:
         """Fetch data."""
         try:
             bbox_info = self.check_list(await self.bbox.device.async_get_bbox_info())
-            devices = self.check_list(await self.bbox.lan.async_get_connected_devices())
+            devices = await self.bbox.lan.async_get_connected_devices()
+            assert isinstance(devices, list), f"Failed to retrieved devices from Bbox API: {devices}"
             wan_ip_stats = self.check_list(await self.bbox.wan.async_get_wan_ip_stats())
             # wan = self.check_list(await self.bbox.wan.async_get_wan_ip())
             # iptv_channels_infos = self.check_list(await self.bbox.iptv.async_get_iptv_info())
@@ -54,7 +55,7 @@ async def _async_update_data(self) -> dict[str, dict[str, Any]]:
 
         return {
             "info": bbox_info,
-            "devices": devices,
+            "devices": self.merge_objects(devices),
             "wan_ip_stats": wan_ip_stats,
             # "wan": wan,
             # "iptv_channels_infos": iptv_channels_infos,
@@ -63,12 +64,38 @@ async def _async_update_data(self) -> dict[str, dict[str, Any]]:
             # "device_info": device_info,
         }
 
+    @staticmethod
+    def merge_objects(objs: Any) -> dict[str, Any]:
+        """Merge objects return by the Bbox API"""
+        assert isinstance(objs, list)
+        def merge(a: dict, b: dict, path=[]):
+            for key in b:
+                if key in a:
+                    if isinstance(a[key], dict) and isinstance(b[key], dict):
+                        merge(a[key], b[key], path + [str(key)])
+                    elif isinstance(a[key], list) and isinstance(b[key], list):
+                        a[key].extend(b[key])
+                    elif a[key] != b[key]:
+                        raise ValueError(
+                            f"Conflict merging the key {'.'.join(path + [str(key)])} of the "
+                            "objects return by the Bbox API: "
+                            f"'{a[key]}' ({type(a[key])}) != '{b[key]}' ({type(b[key])})"
+                        )
+                else:
+                    a[key] = b[key]
+            return a
+        result = objs[0]
+        assert isinstance(result, dict), f"The first element of the list is not a dict (but {type(result)}): {result}"
+        for idx, obj in enumerate(objs[1:]):
+            assert isinstance(obj, dict), f"The {idx+2} element of the list is not a dict (but {type(obj)}): {obj}"
+            result = merge(result, obj)
+        return result
+
     @staticmethod
     def check_list(obj: Any) -> dict[str, Any]:
         """Return element if one only."""
-        if isinstance(obj, list) and len(obj) == 1:
-            return obj[0]
-
-        raise UpdateFailed(
-            "The call is not a list or it contains more than one element"
-        )
+        if not isinstance(obj, list):
+            raise UpdateFailed(f"The call is not a list ({type(obj)}): {obj}")
+        if len(obj) != 1:
+            raise UpdateFailed(f"The call contains more than one element ({len(obj)}): {obj}")
+        return obj[0]

custom_components/bbox/helpers.py

@@ -15,6 +15,7 @@
 class BboxSensorDescription(SensorEntityDescription):
     """Describes a sensor."""
 
+    get_value: Callable[..., Any] | None = None
     value_fn: Callable[..., StateType] | None = None
 
 

custom_components/bbox/sensor.py

@@ -9,7 +9,7 @@
     SensorEntity,
     SensorStateClass,
 )
-from homeassistant.const import PERCENTAGE, UnitOfDataRate, UnitOfTemperature
+from homeassistant.const import UnitOfDataRate, UnitOfInformation, UnitOfTemperature, PERCENTAGE
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
@@ -30,11 +30,11 @@
     ),
     BboxSensorDescription(
         key="wan_ip_stats.wan.ip.stats.rx.bytes",
-        name="Download speed",
-        device_class=SensorDeviceClass.DATA_RATE,
-        native_unit_of_measurement=UnitOfDataRate.KILOBYTES_PER_SECOND,
+        name="Downloaded",
+        device_class=SensorDeviceClass.DATA_SIZE,
+        native_unit_of_measurement=UnitOfInformation.KILOBITS,
         icon="mdi:download-network",
-        value_fn=lambda x: round(x / 1000000, 2),
+        value_fn=lambda x: round(float(x) / 1000, 2),
         state_class=SensorStateClass.MEASUREMENT,
     ),
     BboxSensorDescription(
@@ -48,21 +48,33 @@
         icon="mdi:upload-network",
     ),
     BboxSensorDescription(
-        key="wan_ip_stats.wan.ip.stats.rx.occupation",
-        name="Download bandwidth occupation (%)",
+        key="wan_ip_stats.wan.ip.stats.rx.bandwidth",
+        name="Download speed",
         device_class=SensorDeviceClass.DATA_RATE,
         icon="mdi:upload-network",
         value_fn=lambda x: float(x),
+        native_unit_of_measurement=UnitOfDataRate.KILOBITS_PER_SECOND,
+        state_class=SensorStateClass.MEASUREMENT,
+    ),
+    BboxSensorDescription(
+        key="wan_ip_stats.wan.ip.stats.rx.occupation",
+        name="Download bandwidth occupation",
+        device_class=SensorDeviceClass.POWER_FACTOR,
+        icon="mdi:upload-network",
+        get_value=lambda self: (
+            float(finditem(self.coordinator.data, "wan_ip_stats.wan.ip.stats.rx.bandwidth")) * 100 /
+            float(finditem(self.coordinator.data, "wan_ip_stats.wan.ip.stats.rx.maxBandwidth"))
+        ),
         native_unit_of_measurement=PERCENTAGE,
         state_class=SensorStateClass.MEASUREMENT,
     ),
     BboxSensorDescription(
         key="wan_ip_stats.wan.ip.stats.tx.bytes",
-        name="Upload speed",
-        device_class=SensorDeviceClass.DATA_RATE,
-        native_unit_of_measurement=UnitOfDataRate.KILOBYTES_PER_SECOND,
+        name="Uploaded",
+        device_class=SensorDeviceClass.DATA_SIZE,
+        native_unit_of_measurement=UnitOfInformation.KILOBITS,
         icon="mdi:upload-network",
-        value_fn=lambda x: round(x / 1000000, 2),
+        value_fn=lambda x: round(float(x) / 1000, 2),
         state_class=SensorStateClass.MEASUREMENT,
     ),
     BboxSensorDescription(
@@ -76,11 +88,23 @@
         icon="mdi:upload-network",
     ),
     BboxSensorDescription(
-        key="wan_ip_stats.wan.ip.stats.tx.occupation",
-        name="Upload bandwidth occupation (%)",
+        key="wan_ip_stats.wan.ip.stats.tx.bandwidth",
+        name="Upload speed",
         device_class=SensorDeviceClass.DATA_RATE,
         icon="mdi:upload-network",
         value_fn=lambda x: float(x),
+        native_unit_of_measurement=UnitOfDataRate.KILOBITS_PER_SECOND,
+        state_class=SensorStateClass.MEASUREMENT,
+    ),
+    BboxSensorDescription(
+        key="wan_ip_stats.wan.ip.stats.tx.occupation",
+        name="Upload bandwidth occupation",
+        device_class=SensorDeviceClass.POWER_FACTOR,
+        icon="mdi:upload-network",
+        get_value=lambda self: (
+            float(finditem(self.coordinator.data, "wan_ip_stats.wan.ip.stats.tx.bandwidth")) * 100 /
+            float(finditem(self.coordinator.data, "wan_ip_stats.wan.ip.stats.tx.maxBandwidth"))
+        ),
         native_unit_of_measurement=PERCENTAGE,
         state_class=SensorStateClass.MEASUREMENT,
     ),
@@ -102,7 +126,13 @@ class BboxSensor(BboxEntity, SensorEntity):
     @property
     def native_value(self):
         """Return sensor state."""
-        data = finditem(self.coordinator.data, self.entity_description.key)
-        if self.entity_description.value_fn is not None:
-            return self.entity_description.value_fn(data)
-        return data
+        raw_value = (
+            self.entity_description.get_value(self)
+            if self.entity_description.get_value is not None
+            else finditem(self.coordinator.data, self.entity_description.key)
+        )
+        return (
+            self.entity_description.value_fn(raw_value)
+            if self.entity_description.value_fn is not None
+            else raw_value
+        )