Add diagnostic information to the device

This encapsulated diagnostic information entirely within the device since the device has all of the local information such as traits, device information, product information, etc. The motivation is to facilitate local testing rather than testing via Home Assistant and dealing with mocks and it could be helpful for other users of the library as well.

Author: allenporter

roborock/devices/device.py

@@ -6,7 +6,8 @@
 
 import logging
 from abc import ABC
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
+from typing import Any, TypeVar, cast
 
 from roborock.data import HomeDataDevice, HomeDataProduct
 from roborock.roborock_message import RoborockMessage
@@ -113,3 +114,41 @@ async def close(self) -> None:
     def _on_message(self, message: RoborockMessage) -> None:
         """Handle incoming messages from the device."""
         _LOGGER.debug("Received message from device: %s", message)
+
+    def diagnostic_data(self) -> dict[str, Any]:
+        """Return diagnostics information about the device."""
+        extra: dict[str, Any]= {}
+        if self.v1_properties:
+            extra["traits"] = _redact_data(self.v1_properties.as_dict())
+        return {
+                "device": _redact_data(self.device_info.as_dict()),
+                "product": _redact_data(self.product.as_dict()),
+                **extra,
+        }
+
+
+T = TypeVar("T")
+
+REDACT_KEYS = {"duid", "localKey", "mac", "bssid"}
+REDACTED = "**REDACTED**"
+
+
+def _redact_data(data: T) -> T | dict[str, Any] | list[Any]:
+    """Redact sensitive data in a dict."""
+    if not isinstance(data, (Mapping, list)):
+        return data
+
+    if isinstance(data, list):
+        return cast(T, [_redact_data(item) for item in data])
+
+    redacted = {**data}
+
+    for key, value in redacted.items():
+        if key in REDACT_KEYS:
+            redacted[key] = REDACTED
+        elif isinstance(value, dict):
+            redacted[key] = _redact_data(value)
+        elif isinstance(value, list):
+            redacted[key] = [_redact_data(item) for item in value]
+
+    return redacted

roborock/devices/traits/v1/__init__.py

@@ -34,7 +34,7 @@
 from dataclasses import dataclass, field, fields
 from typing import Any, get_args
 
-from roborock.data.containers import HomeData, HomeDataProduct
+from roborock.data.containers import HomeData, HomeDataProduct, RoborockBase
 from roborock.data.v1.v1_code_mappings import RoborockDockTypeCode
 from roborock.devices.cache import Cache
 from roborock.devices.traits import Trait
@@ -247,6 +247,16 @@ async def _set_cached_trait_data(self, name: str, value: Any) -> None:
         _LOGGER.debug("Updating cached trait data: %s", cache_data.trait_data)
         await self._cache.set(cache_data)
 
+    def as_dict(self) -> dict[str, Any]:
+        """Return the trait data as a dictionary."""
+        result: dict[str, Any] = {}
+        for item in fields(self):
+            trait = getattr(self, item.name, None)
+            if trait is None or not isinstance(trait, RoborockBase):
+                continue
+            result[item.name] = trait.as_dict()
+        return result
+
 
 def create(
     device_uid: str,

roborock/devices/traits/v1/device_features.py

@@ -1,3 +1,4 @@
+from dataclasses import fields
 from typing import Self
 
 from roborock.data import AppInitStatus, RoborockProductNickname
@@ -16,6 +17,10 @@ def __init__(self, product_nickname: RoborockProductNickname, cache: Cache) -> N
         """Initialize MapContentTrait."""
         self._nickname = product_nickname
         self._cache = cache
+        # All fields of DeviceFeatures are required. Initialize them to False
+        # so we have some known state.
+        for field in fields(self):
+            setattr(self, field.name, False)
 
     async def refresh(self) -> Self:
         """Refresh the contents of this trait.

roborock/devices/traits/v1/network_info.py

@@ -29,6 +29,7 @@ def __init__(self, device_uid: str, cache: Cache) -> None:
         """Initialize the trait."""
         self._device_uid = device_uid
         self._cache = cache
+        self.ip = ""
 
     async def refresh(self) -> Self:
         """Refresh the network info from the cache."""