Fabric management (#249)

Author: ArturoGuerra

matter_server/client/client.py

@@ -8,6 +8,7 @@
 import uuid
 
 from aiohttp import ClientSession
+from chip.clusters import Objects as Clusters
 
 from matter_server.common.errors import ERROR_MAP
 
@@ -27,7 +28,7 @@
 )
 from .connection import MatterClientConnection
 from .exceptions import ConnectionClosed, InvalidServerVersion, InvalidState
-from .models.node import MatterNode
+from .models.node import MatterFabricData, MatterNode
 
 if TYPE_CHECKING:
     from chip.clusters.Objects import ClusterCommand
@@ -156,6 +157,46 @@ async def open_commissioning_window(
             ),
         )
 
+    async def get_matter_fabrics(self, node_id: int) -> list[MatterFabricData]:
+        """
+        Get Matter fabrics from a device.
+
+        Returns a list of MatterFabricData objects.
+        """
+
+        node = await self.get_node(node_id)
+        fabrics: list[
+            Clusters.OperationalCredentials.Structs.FabricDescriptor
+        ] = node.get_attribute_value(
+            0, None, Clusters.OperationalCredentials.Attributes.Fabrics
+        )
+
+        vendors_map = await self.send_command(
+            APICommand.GET_VENDOR_NAMES,
+            filter_vendors=[f.vendorId for f in fabrics],
+        )
+
+        return [
+            MatterFabricData(
+                fabric_id=f.fabricId,
+                vendor_id=f.vendorId,
+                fabric_index=f.fabricIndex,
+                fabric_label=f.label if f.label else None,
+                vendor_name=vendors_map.get(f.vendorId),
+            )
+            for f in fabrics
+        ]
+
+    async def remove_matter_fabric(self, node_id: int, fabric_index: int) -> None:
+        """Remove Matter fabric from a device."""
+        await self.send_device_command(
+            node_id,
+            0,
+            Clusters.OperationalCredentials.Commands.RemoveFabric(
+                fabricIndex=fabric_index,
+            ),
+        )
+
     async def send_device_command(
         self,
         node_id: int,

matter_server/client/models/node.py

@@ -1,6 +1,7 @@
 """Matter node."""
 from __future__ import annotations
 
+from dataclasses import dataclass
 import logging
 from typing import Any, TypeVar, cast
 
@@ -40,6 +41,17 @@ def get_object_params(
     raise KeyError(f"No descriptor found for object {object_id}")
 
 
+@dataclass
+class MatterFabricData:
+    """Data about a Matter fabric."""
+
+    fabric_id: int
+    vendor_id: int
+    fabric_index: int
+    fabric_label: str | None = None
+    vendor_name: str | None = None
+
+
 class MatterEndpoint:
     """Representation of a Matter Endpoint."""
 

matter_server/common/models.py

@@ -37,13 +37,26 @@ class APICommand(str, Enum):
     INTERVIEW_NODE = "interview_node"
     DEVICE_COMMAND = "device_command"
     REMOVE_NODE = "remove_node"
+    GET_VENDOR_NAMES = "get_vendor_names"
 
 
 EventCallBackType = Callable[[EventType, Any], None]
 
 # Generic model(s)
 
 
+@dataclass
+class VendorInfo:
+    """Vendor info as received from the CSA."""
+
+    vendor_id: int
+    vendor_name: str
+    company_legal_name: str
+    company_preferred_name: str
+    vendor_landing_page_url: str
+    creator: str
+
+
 @dataclass
 class MatterNodeData:
     """Matter node data as received from (and stored on) the server."""

matter_server/server/server.py

@@ -25,6 +25,7 @@
 from .device_controller import MatterDeviceController
 from .stack import MatterStack
 from .storage import StorageController
+from .vendor_info import VendorInfo
 
 
 def mount_websocket(server: MatterServer, path: str) -> None:
@@ -75,6 +76,7 @@ def __init__(
         # of Matter devices and their subscriptions.
         self.device_controller = MatterDeviceController(self)
         self.storage = StorageController(self)
+        self.vendor_info = VendorInfo(self)
         # we dynamically register command handlers
         self.command_handlers: dict[str, APICommandHandler] = {}
         self._subscribers: Set[EventCallBackType] = set()
@@ -92,6 +94,7 @@ async def start(self) -> None:
         await self.device_controller.initialize()
         await self.storage.start()
         await self.device_controller.start()
+        await self.vendor_info.start()
         mount_websocket(self, "/ws")
         self.app.router.add_route("GET", "/", self._handle_info)
         self._runner = web.AppRunner(self.app, access_log=None)
@@ -174,7 +177,7 @@ def register_api_command(
 
     def _register_api_commands(self) -> None:
         """Register all methods decorated as api_command."""
-        for cls in (self, self.device_controller):
+        for cls in (self, self.device_controller, self.vendor_info):
             for attr_name in dir(cls):
                 if attr_name.startswith("__"):
                     continue

matter_server/server/vendor_info.py

@@ -0,0 +1,97 @@
+"""Fetches vendor info from the CSA."""
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from aiohttp import ClientError, ClientSession
+
+from ..common.helpers.api import api_command
+from ..common.helpers.util import dataclass_from_dict, dataclass_to_dict
+from ..common.models import APICommand, VendorInfo as VendorInfoModel
+
+if TYPE_CHECKING:
+    from .server import MatterServer
+
+LOGGER = logging.getLogger(__name__)
+PRODUCTION_URL = "https://on.dcl.csa-iot.org"
+DATA_KEY_VENDOR_INFO = "vendor_info"
+
+
+class VendorInfo:
+    """Fetches vendor info from the CSA and handles api calls to get it."""
+
+    def __init__(self, server: MatterServer):
+        """Initialize the vendor info."""
+        self._data: dict[int, VendorInfoModel] = {}
+        self._server = server
+
+    async def start(self) -> None:
+        """Async initialize the vendor info."""
+        self._load_vendors()
+        await self._fetch_vendors()
+        self._save_vendors()
+
+    def _load_vendors(self) -> None:
+        """Load vendor info from storage."""
+        LOGGER.info("Loading vendor info from storage.")
+        vendor_count = 0
+        data = self._server.storage.get(DATA_KEY_VENDOR_INFO, {})
+        for vendor_id, vendor_info in data.items():
+            self._data[vendor_id] = dataclass_from_dict(VendorInfoModel, vendor_info)
+            vendor_count += 1
+        LOGGER.info("Loaded %s vendors from storage.", vendor_count)
+
+    async def _fetch_vendors(self) -> None:
+        """Fetch the vendor names from the CSA."""
+        LOGGER.info("Fetching the latest vendor info from DCL.")
+        vendors: dict[int, VendorInfoModel] = {}
+        try:
+            async with ClientSession(raise_for_status=True) as session:
+                async with session.get(
+                    f"{PRODUCTION_URL}/dcl/vendorinfo/vendors"
+                ) as response:
+                    data = await response.json()
+                    for vendorinfo in data["vendorInfo"]:
+                        vendors[vendorinfo["vendorID"]] = VendorInfoModel(
+                            vendor_id=vendorinfo["vendorID"],
+                            vendor_name=vendorinfo["vendorName"],
+                            company_legal_name=vendorinfo["companyLegalName"],
+                            company_preferred_name=vendorinfo["companyPreferredName"],
+                            vendor_landing_page_url=vendorinfo["vendorLandingPageURL"],
+                            creator=vendorinfo["creator"],
+                        )
+        except ClientError as err:
+            LOGGER.error("Unable to fetch vendor info from DCL: %s", err)
+        else:
+            LOGGER.info("Fetched %s vendors from DCL.", len(vendors))
+
+        self._data.update(vendors)
+
+    def _save_vendors(self) -> None:
+        """Save vendor info to storage."""
+        LOGGER.info("Saving vendor info to storage.")
+        self._server.storage.set(
+            DATA_KEY_VENDOR_INFO,
+            {
+                vendor_id: dataclass_to_dict(vendor_info)
+                for vendor_id, vendor_info in self._data.items()
+            },
+        )
+
+    @api_command(APICommand.GET_VENDOR_NAMES)
+    async def get_vendor_names(
+        self, filter_vendors: list[int] | None = None
+    ) -> dict[int, str]:
+        """Get a map of vendor ids to vendor names."""
+        if filter_vendors:
+            vendors: dict[int, str] = {}
+            for vendor_id in filter_vendors:
+                if vendor_id in filter_vendors:
+                    vendors[vendor_id] = self._data[vendor_id].vendor_name
+            return vendors
+
+        return {
+            vendor_id: vendor_info.vendor_name
+            for vendor_id, vendor_info in self._data.items()
+        }