Add show_stored_node script

.github/workflows/test.yml

@@ -29,10 +29,12 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install -r requirements-test.txt
+          pip install -e . -r requirements-test.txt homeassistant
       - name: Flake8
-        run: flake8 custom_components matter_server/client matter_server/common matter_server/server
+        run: flake8 scripts custom_components matter_server/client matter_server/common matter_server/server
       - name: Black
-        run: black --check custom_components matter_server/client matter_server/common matter_server/server
+        run: black --check scripts custom_components matter_server/client matter_server/common matter_server/server
       - name: isort
-        run: isort --check custom_components matter_server/client matter_server/common matter_server/server
+        run: isort --check scripts custom_components matter_server/client matter_server/common matter_server/server
+      - name: test show stored node script
+        run: python3 -m scripts.show_stored_node tests/fixtures/nodes/lighting-example-app.json

matter_server/client/model/device.py

@@ -165,3 +165,6 @@ def _receive_event(self, event):
 
         if self._on_update_listener:
             self._on_update_listener()
+
+    def __repr__(self):
+        return f"<MatterDevice {self.device_type.__name__} (N:{self.node.node_id}, E:{self.endpoint_id})>"

scripts/chip_synchronize.py

@@ -1,30 +1,37 @@
 import pathlib
+
 import git
 
 REPO_ROOT = pathlib.Path(__file__).parent.parent
 
 CHIP_ROOT = REPO_ROOT / "../connectedhomeip"
 
 CLUSTER_OBJECTS = REPO_ROOT / "matter_server/vendor/chip/clusters/Objects.py"
-CLUSTER_OBJECTS_VERSION = REPO_ROOT / "matter_server/vendor/chip/clusters/ObjectsVersion.py"
+CLUSTER_OBJECTS_VERSION = (
+    REPO_ROOT / "matter_server/vendor/chip/clusters/ObjectsVersion.py"
+)
 
 REPLACE_IMPORT = {
     "from chip import ChipUtility\n": "from .. import ChipUtility\n",
-    "from chip.tlv import uint, float32\n": "from ..tlv import uint, float32\n"
+    "from chip.tlv import uint, float32\n": "from ..tlv import uint, float32\n",
 }
 
 
 def main():
     repo = git.Repo(CHIP_ROOT)
 
     with open(CLUSTER_OBJECTS, "w") as cluster_outfile:
-        with open(CHIP_ROOT / "src/controller/python/chip/clusters/Objects.py", "r") as cluster_file:
+        with open(
+            CHIP_ROOT / "src/controller/python/chip/clusters/Objects.py", "r"
+        ) as cluster_file:
             for line in cluster_file:
                 if line in REPLACE_IMPORT.keys():
                     line = REPLACE_IMPORT[line]
                 cluster_outfile.write(line)
     with open(CLUSTER_OBJECTS_VERSION, "w") as cluster_version_outfile:
-        cluster_version_outfile.write(f"CLUSTER_OBJECT_VERSION = \"{repo.head.object.hexsha}\"")
+        cluster_version_outfile.write(
+            f'CLUSTER_OBJECT_VERSION = "{repo.head.object.hexsha}"'
+        )
 
 
 if __name__ == "__main__":

scripts/list_unmapped_devices.py

@@ -5,7 +5,6 @@
 from custom_components.matter_experimental.device_platform import DEVICE_PLATFORM
 from matter_server.vendor import device_types
 
-
 IGNORE_DEVICES = {
     device_types.AllClustersAppServerExample,
     device_types.Bridge,

scripts/show_stored_node.py

@@ -0,0 +1,103 @@
+"""Show mappings for given JSON."""
+
+import dataclasses
+import json
+import logging
+import os
+import pathlib
+import sys
+from unittest.mock import Mock
+
+from custom_components.matter_experimental.device_platform import DEVICE_PLATFORM
+from matter_server.client.model.device import MatterDevice
+from matter_server.client.model.node import MatterNode
+from matter_server.common import json_utils
+
+
+def resolve_input():
+    if len(sys.argv) < 2:
+        print("Pass in path to JSON file containing single node or HA storage")
+        sys.exit(1)
+
+    path = sys.argv[1]
+
+    return (pathlib.Path(os.getcwd()) / path).read_text()
+
+
+def print_node(node: MatterNode):
+    print(node)
+    first = True
+    for device in node.devices:
+        if first:
+            first = False
+        else:
+            print()
+        print_device(device)
+
+
+def print_device(device: MatterDevice):
+    created = False
+    print(f"  {device}")
+
+    for platform, devices in DEVICE_PLATFORM.items():
+        device_mappings = devices.get(device.device_type)
+
+        if device_mappings is None:
+            continue
+
+        if not isinstance(device_mappings, list):
+            device_mappings = [device_mappings]
+
+        for device_mapping in device_mappings:
+            created = True
+            print(f"    - Platform: {platform}")
+
+            for key, value in sorted(dataclasses.asdict(device_mapping).items()):
+                if value is None:
+                    continue
+                if key == "entity_cls":
+                    value = value.__name__
+
+                if key != "subscribe_attributes":
+                    print(f"      {key}: {value}")
+                    continue
+
+                print("      Subscriptions:")
+
+                for sub in value:
+                    print(f"       - {sub.__qualname__}")
+
+                # Try instantiating to ensure the device mapping doesn't crash
+                device_mapping.entity_cls(device, device_mapping)
+
+    # Do not warng on root node
+    if not created:
+        print("  ** WARNING: NOT MAPPED IN HOME ASSISTANT")
+
+
+def main():
+    raw_data = resolve_input()
+    data = json.loads(raw_data, cls=json_utils.CHIPJSONDecoder)
+
+    # This is a HA storage file. Extract nodes
+    if "key" in data and data["key"].startswith("matter_experimental_"):
+        nodes = [d for d in data["data"]["nodes"].values() if d is not None]
+    else:
+        nodes = [data]
+
+    first = True
+
+    mock_matter = Mock(adapter=Mock(logger=logging.getLogger("show_mappings")))
+
+    for node_data in nodes:
+        if first:
+            first = False
+        else:
+            print()
+            print()
+
+        print_node(MatterNode(mock_matter, node_data))
+
+
+if __name__ == "__main__":
+    main()