tessl/pypi-home-assistant-chip-clusters

Python APIs and tools for Matter (Project CHIP) protocol implementation, specifically the chip clusters functionality used by Home Assistant for Matter device control and communication

—

Pending

Overview

Eval results

Files

Device Controller

Name: tessl/pypi-home-assistant-chip-clusters
Author: tessl

Core device controller functionality for Matter fabric management, device commissioning, and communication. The ChipDeviceController class provides the primary interface for interacting with Matter devices.

Capabilities

Device Controller Initialization

Creates and configures a device controller instance for managing Matter devices within a fabric.

class ChipDeviceController:
    def __init__(
        self,
        opCredsContext: ctypes.c_void_p,
        fabricId: int,
        nodeId: int,
        adminVendorId: int,
        catTags: list = [],
        paaTrustStorePath: str = "",
        useTestCommissioner: bool = False,
        fabricAdmin: FabricAdmin = None,
        name: str = "",
        keypair: P256Keypair = None
    ):
        """
        Initialize a CHIP Device Controller.

        Parameters:
        - opCredsContext: Operational credentials context
        - fabricId: Fabric ID for this controller
        - nodeId: Node ID for this controller
        - adminVendorId: Administrator vendor ID
        - catTags: List of CASE Authenticated Tags
        - paaTrustStorePath: Path to PAA trust store certificates
        - useTestCommissioner: Whether to use test commissioner credentials
        - fabricAdmin: Fabric administrator instance
        - name: Name for this controller
        - keypair: Optional P256 keypair for this controller
        """

Device Commissioning

Commission new Matter devices into the fabric using network-based or code-based methods.

async def CommissionOnNetwork(
    self,
    nodeId: int,
    setupPinCode: int,
    filterType: int = None,
    filter: str = None,
    discoveryTimeoutMsec: int = 30000
) -> bool:
    """
    Commission a device over the network.

    Parameters:
    - nodeId: Target node ID for the device
    - setupPinCode: Device setup PIN code
    - filterType: Discovery filter type (0=None, 1=Short Discriminator, 2=Long Discriminator, 3=Vendor ID, 4=Device Type)
    - filter: Filter value based on filterType
    - discoveryTimeoutMsec: Discovery timeout in milliseconds

    Returns:
    True if commissioning started successfully
    """

async def CommissionWithCode(
    self,
    setupPayload: str,
    nodeId: int,
    discoveryType: int = 0
) -> bool:
    """
    Commission a device using setup payload (QR code or manual code).

    Parameters:
    - setupPayload: Setup payload string from QR code or manual entry
    - nodeId: Target node ID for the device
    - discoveryType: Discovery method (0=Network Only, 1=BLE Only, 2=On Network, 3=BLE WiFi, 4=BLE Thread)

    Returns:
    True if commissioning started successfully
    """

Device Connection Management

Manage connections to commissioned devices and retrieve device proxy objects.

def GetConnectedDevice(
    self,
    nodeId: int,
    allowPASE: bool = False,
    timeoutMs: int = None
):
    """
    Get a connected device proxy for the specified node.

    Parameters:
    - nodeId: Node ID of the target device
    - allowPASE: Allow PASE (password-authenticated session establishment) connections
    - timeoutMs: Connection timeout in milliseconds

    Returns:
    Device proxy object for interacting with the device
    """

def IsConnected(self, nodeId: int) -> bool:
    """
    Check if a device is currently connected.

    Parameters:
    - nodeId: Node ID to check

    Returns:
    True if device is connected
    """

Attribute Operations

Read, write, and subscribe to device attributes using the Matter interaction model.

async def ReadAttribute(
    self,
    nodeid: int,
    attributes: list,
    fabricFiltered: bool = True,
    returnRawTLV: bool = False,
    suppressResponse: bool = False,
    reportInterval: tuple = None,
    keepSubscriptions: bool = False,
    autoResubscribe: bool = True
):
    """
    Read attributes from a device.

    Parameters:
    - nodeid: Target device node ID
    - attributes: List of (endpoint, attribute) tuples or AttributePath objects
    - fabricFiltered: Filter attributes by current fabric
    - returnRawTLV: Return raw TLV data instead of decoded values
    - suppressResponse: Suppress response for write operations
    - reportInterval: Tuple of (min_interval, max_interval) for subscriptions
    - keepSubscriptions: Keep existing subscriptions when creating new ones
    - autoResubscribe: Automatically resubscribe on connection loss

    Returns:
    Dictionary of attribute values keyed by (endpoint, cluster, attribute)
    """

async def WriteAttribute(
    self,
    nodeid: int,
    attributes: list,
    timedRequestTimeoutMs: int = None,
    interactionTimeoutMs: int = None,
    suppressResponse: bool = False
):
    """
    Write attributes to a device.

    Parameters:
    - nodeid: Target device node ID
    - attributes: List of (endpoint, attribute, value) tuples
    - timedRequestTimeoutMs: Timeout for timed requests
    - interactionTimeoutMs: Overall interaction timeout
    - suppressResponse: Suppress write response

    Returns:
    Write response status
    """

async def ReadAttribute(
    self,
    nodeid: int,
    attributes: list,
    reportInterval: tuple = None,
    keepSubscriptions: bool = False,
    autoResubscribe: bool = True,
    fabricFiltered: bool = True,
    **kwargs
):
    """
    Read attributes from a device. When reportInterval is provided, creates a subscription instead of a one-time read.

    Parameters:
    - nodeid: Target device node ID
    - attributes: List of (endpoint, attribute) tuples or AttributePath objects
    - reportInterval: Tuple of (min_interval, max_interval) in seconds. When provided, creates subscription.
    - keepSubscriptions: Keep existing subscriptions when creating new ones
    - autoResubscribe: Automatically resubscribe on connection loss
    - fabricFiltered: Filter attributes by current fabric

    Returns:
    Dictionary of attribute values for one-time read, or SubscriptionTransaction object for subscriptions
    """

Command Operations

Send cluster commands to devices and handle responses.

async def SendCommand(
    self,
    nodeid: int,
    endpoint: int,
    command,
    timedRequestTimeoutMs: int = None,
    interactionTimeoutMs: int = None,
    busyWaitMs: int = None,
    suppressResponse: bool = False,
    remotePassiveTimeout: int = None,
    suppressTimedRequestMessage: bool = False,
    responseCallback=None
):
    """
    Send a cluster command to a device.

    Parameters:
    - nodeid: Target device node ID
    - endpoint: Target endpoint ID
    - command: Cluster command object
    - timedRequestTimeoutMs: Timeout for timed requests
    - interactionTimeoutMs: Overall interaction timeout
    - busyWaitMs: Busy wait timeout
    - suppressResponse: Suppress command response
    - remotePassiveTimeout: Remote passive timeout
    - suppressTimedRequestMessage: Suppress timed request message
    - responseCallback: Callback for command response

    Returns:
    Command response or None if suppressResponse=True
    """

def SendGroupCommand(
    self,
    groupid: int,
    command,
    busyWaitMs: int = None
):
    """
    Send a command to a group of devices.

    Parameters:
    - groupid: Target group ID
    - command: Cluster command object
    - busyWaitMs: Busy wait timeout

    Returns:
    Command response
    """

Device Management

Additional device management operations including fabric and group management.

def RemoveDevice(self, nodeid: int):
    """
    Remove a device from the fabric.

    Parameters:
    - nodeid: Node ID of device to remove
    """

def SetWiFiCredentials(self, ssid: str, credentials: str):
    """
    Set WiFi credentials for device commissioning.

    Parameters:
    - ssid: WiFi network SSID
    - credentials: WiFi password
    """

def SetThreadOperationalDataset(self, threadOperationalDataset: bytes):
    """
    Set Thread operational dataset for device commissioning.

    Parameters:
    - threadOperationalDataset: Thread network dataset
    """

def Shutdown(self):
    """
    Clean shutdown of the device controller.
    Closes all connections and releases resources.
    """

Usage Examples

Basic Device Control

import chip.ChipDeviceCtrl
from chip.ChipDeviceCtrl import ChipDeviceController
from chip.FabricAdmin import FabricAdmin
import chip.clusters as Clusters
import asyncio

async def main():
    # Initialize fabric admin and controller
    fabricAdmin = FabricAdmin()
    controller = fabricAdmin.NewController(
        nodeId=12345,
        paaTrustStorePath=None,
        useTestCommissioner=False
    )

    # Commission a device
    await controller.CommissionOnNetwork(
        nodeId=1,
        setupPinCode=20202021,
        discoveryTimeoutMsec=30000
    )

    # Read device basic information
    basic_info = await controller.ReadAttribute(
        nodeid=1,
        attributes=[(0, Clusters.BasicInformation.Attributes.VendorName),
                   (0, Clusters.BasicInformation.Attributes.ProductName)]
    )
    
    # Control an OnOff device
    await controller.SendCommand(
        nodeid=1,
        endpoint=1,
        Clusters.OnOff.Commands.On()
    )

    # Clean shutdown
    controller.Shutdown()

# Run the async function
asyncio.run(main())

Subscription Handling

# Subscribe to attribute changes using ReadAttribute with reportInterval
subscription = await controller.ReadAttribute(
    nodeid=1,
    attributes=[(1, Clusters.OnOff.Attributes.OnOff),
               (1, Clusters.LevelControl.Attributes.CurrentLevel)],
    reportInterval=(1, 10)  # Report changes every 1-10 seconds
)

# Set up callbacks to handle subscription updates
def attribute_update_callback(path, transaction):
    print(f"Attribute updated: {path}")

subscription.SetAttributeUpdateCallback(attribute_update_callback)

Install with Tessl CLI

npx tessl i tessl/pypi-home-assistant-chip-clusters

docs

ble-discovery.md

clusters.md

crypto-credentials.md