Add pending dataset visibility and cancellation to OTBR WebSocket API

homeassistant/components/otbr/util.py

@@ -152,6 +152,57 @@ async def set_channel(
         """Set current channel."""
         await self.api.set_channel(channel, delay=int(delay * 1000))
 
+    async def delete_pending_dataset(self) -> None:
+        """Delete the pending operational dataset.
+
+        Tries DELETE first (spec-compliant). Falls back to creating a pending
+        dataset with the current active dataset and delay=0, which immediately
+        applies the current state and clears the pending migration.
+        """
+        # Try the spec-compliant DELETE first; if it succeeds, return early.
+        # If it fails (e.g. 405 on older firmware), fall through to the
+        # fallback which overwrites the pending dataset instead.
+        try:
+            await self.api.delete_pending_dataset()
+        except (python_otbr_api.OTBRError, aiohttp.ClientError, TimeoutError) as exc:
+            _LOGGER.debug(
+                "DELETE pending dataset not supported, using fallback: %s", exc
+            )
+        else:
+            return
+
+        # Fallback: set a pending dataset matching the current active dataset
+        # with delay=0, which immediately applies and clears the pending state.
+        try:
+            dataset = await self.api.get_active_dataset()
+        except (
+            python_otbr_api.OTBRError,
+            aiohttp.ClientError,
+            TimeoutError,
+        ) as exc:
+            raise HomeAssistantError("Failed to call OTBR API") from exc
+
+        if dataset is None:
+            raise HomeAssistantError(
+                "Failed to cancel pending dataset: no active dataset"
+            )
+
+        if dataset.active_timestamp and dataset.active_timestamp.seconds is not None:
+            dataset.active_timestamp.seconds += 1
+        else:
+            dataset.active_timestamp = python_otbr_api.Timestamp(False, 1, 0)
+
+        try:
+            await self.api.create_pending_dataset(
+                python_otbr_api.PendingDataSet(active_dataset=dataset, delay=0)
+            )
+        except (
+            python_otbr_api.OTBRError,
+            aiohttp.ClientError,
+            TimeoutError,
+        ) as exc:
+            raise HomeAssistantError("Failed to call OTBR API") from exc
+
     @_handle_otbr_error
     async def get_extended_address(self) -> bytes:
         """Get extended address (EUI-64)."""

homeassistant/components/otbr/websocket_api.py

@@ -35,6 +35,8 @@ def async_setup(hass: HomeAssistant) -> None:
     """Set up the OTBR Websocket API."""
     websocket_api.async_register_command(hass, websocket_info)
     websocket_api.async_register_command(hass, websocket_create_network)
+    websocket_api.async_register_command(hass, websocket_get_pending_dataset)
+    websocket_api.async_register_command(hass, websocket_delete_pending_dataset)
     websocket_api.async_register_command(hass, websocket_set_channel)
     websocket_api.async_register_command(hass, websocket_set_network)
 
@@ -295,3 +297,75 @@ async def websocket_set_channel(
         return
 
     connection.send_result(msg["id"], {"delay": delay})
+
+
+@websocket_api.websocket_command(
+    {
+        "type": "otbr/get_pending_dataset",
+        vol.Required("extended_address"): str,
+    }
+)
+@websocket_api.require_admin
+@websocket_api.async_response
+@async_get_otbr_data
+async def websocket_get_pending_dataset(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict,
+    data: OTBRData,
+) -> None:
+    """Get pending dataset info."""
+    try:
+        pending_tlvs = await data.get_pending_dataset_tlvs()
+    except HomeAssistantError as exc:
+        connection.send_error(msg["id"], "get_pending_dataset_failed", str(exc))
+        return
+
+    if pending_tlvs is None:
+        connection.send_result(msg["id"], None)
+        return
+
+    dataset = tlv_parser.parse_tlv(pending_tlvs.hex())
+
+    channel_tlv = dataset.get(MeshcopTLVType.CHANNEL)
+    delay_tlv = dataset.get(MeshcopTLVType.DELAYTIMER)
+
+    if channel_tlv is None or delay_tlv is None:
+        connection.send_result(msg["id"], None)
+        return
+
+    pending_channel = cast(tlv_parser.Channel, channel_tlv).channel
+    delay_ms = int.from_bytes(delay_tlv.data, "big")
+
+    connection.send_result(
+        msg["id"],
+        {
+            "pending_channel": pending_channel,
+            "pending_dataset_delay": delay_ms / 1000,
+        },
+    )
+
+
+@websocket_api.websocket_command(
+    {
+        "type": "otbr/delete_pending_dataset",
+        vol.Required("extended_address"): str,
+    }
+)
+@websocket_api.require_admin
+@websocket_api.async_response
+@async_get_otbr_data
+async def websocket_delete_pending_dataset(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict,
+    data: OTBRData,
+) -> None:
+    """Delete pending dataset."""
+    try:
+        await data.delete_pending_dataset()
+    except HomeAssistantError as exc:
+        connection.send_error(msg["id"], "delete_pending_dataset_failed", str(exc))
+        return
+
+    connection.send_result(msg["id"])