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
—
Bluetooth Low Energy connectivity and device discovery functionality for finding and connecting to Matter devices during commissioning and setup.
Manages Bluetooth Low Energy connections for Matter device discovery and commissioning.
class BleManager:
"""BLE manager for Matter device connectivity."""
def __init__(self):
"""Initialize BLE manager."""
...
def scan(self, timeout: int = 10) -> list:
"""
Scan for BLE devices advertising Matter services.
Parameters:
- timeout: Scan timeout in seconds
Returns:
List of discovered device information dictionaries
"""
...
def connect(self, identifier: str) -> bool:
"""
Connect to a specific BLE device.
Parameters:
- identifier: Device identifier (MAC address or UUID)
Returns:
True if connection successful
"""
...
def disconnect(self) -> bool:
"""
Disconnect from current BLE device.
Returns:
True if disconnection successful
"""
...
def get_connected_device(self):
"""
Get currently connected BLE device.
Returns:
Connected device object or None
"""
...Platform-specific implementations for different operating systems.
class ChipBluezMgr:
"""Linux BlueZ BLE manager for CHIP operations."""
def __init__(self, adapter_name: str = None):
"""
Initialize BlueZ manager.
Parameters:
- adapter_name: Bluetooth adapter name (default: first available)
"""
...
def start_scan(self) -> bool:
"""Start BLE device scanning."""
...
def stop_scan(self) -> bool:
"""Stop BLE device scanning."""
...
def get_peripheral(self, identifier: str):
"""Get peripheral device by identifier."""
...
def connect_peripheral(self, peripheral) -> bool:
"""Connect to a peripheral device."""
...
def disconnect_peripheral(self, peripheral) -> bool:
"""Disconnect from a peripheral device."""
...
class ChipCoreBluetoothMgr:
"""macOS Core Bluetooth manager for CHIP operations."""
def __init__(self):
"""Initialize Core Bluetooth manager."""
...
def start_scan(self, service_uuids: list = None) -> bool:
"""
Start scanning for peripherals.
Parameters:
- service_uuids: List of service UUIDs to filter (optional)
Returns:
True if scan started successfully
"""
...
def stop_scan(self) -> bool:
"""Stop peripheral scanning."""
...
def connect_peripheral(self, peripheral_id: str) -> bool:
"""
Connect to peripheral by identifier.
Parameters:
- peripheral_id: Peripheral identifier
Returns:
True if connection initiated successfully
"""
...
def disconnect_peripheral(self, peripheral_id: str) -> bool:
"""
Disconnect from peripheral.
Parameters:
- peripheral_id: Peripheral identifier
Returns:
True if disconnection initiated successfully
"""
...Discover Matter devices on the network using various discovery methods.
class DiscoveryManager:
"""Manager for Matter device discovery."""
def __init__(self):
"""Initialize discovery manager."""
...
def discover_commissionable_nodes(
self,
timeout: int = 30,
filter_type: int = None,
filter_value: str = None
) -> list:
"""
Discover commissionable Matter nodes on the network.
Parameters:
- timeout: Discovery timeout in seconds
- filter_type: Discovery filter type (0=None, 1=Short Discriminator, 2=Long Discriminator, 3=Vendor ID, 4=Device Type)
- filter_value: Filter value based on filter_type
Returns:
List of discovered node information dictionaries
"""
...
def discover_operational_devices(
self,
fabric_id: int = None,
timeout: int = 30
) -> list:
"""
Discover operational (commissioned) Matter devices.
Parameters:
- fabric_id: Fabric ID to filter by (optional)
- timeout: Discovery timeout in seconds
Returns:
List of operational device information dictionaries
"""
...
def resolve_device(self, node_id: int, fabric_id: int = None) -> dict:
"""
Resolve specific device information.
Parameters:
- node_id: Target node ID
- fabric_id: Fabric ID (optional)
Returns:
Device information dictionary or None if not found
"""
...
def stop_discovery(self):
"""Stop all active discovery operations."""
...Specialized controller for managing commissionable nodes during device setup.
class ChipCommissionableNodeCtrl:
"""Controller for commissionable node operations."""
def __init__(self):
"""Initialize commissionable node controller."""
...
def discover_commissionable_nodes_long_discriminator(
self,
long_discriminator: int,
timeout: int = 30
) -> list:
"""
Discover nodes by long discriminator.
Parameters:
- long_discriminator: Long discriminator value (0-4095)
- timeout: Discovery timeout in seconds
Returns:
List of matching node information dictionaries
"""
...
def discover_commissionable_nodes_short_discriminator(
self,
short_discriminator: int,
timeout: int = 30
) -> list:
"""
Discover nodes by short discriminator.
Parameters:
- short_discriminator: Short discriminator value (0-15)
- timeout: Discovery timeout in seconds
Returns:
List of matching node information dictionaries
"""
...
def discover_commissionable_nodes_vendor_id(
self,
vendor_id: int,
timeout: int = 30
) -> list:
"""
Discover nodes by vendor ID.
Parameters:
- vendor_id: Vendor ID to search for
- timeout: Discovery timeout in seconds
Returns:
List of matching node information dictionaries
"""
...
def discover_commissionable_nodes_device_type(
self,
device_type: int,
timeout: int = 30
) -> list:
"""
Discover nodes by device type.
Parameters:
- device_type: Device type ID to search for
- timeout: Discovery timeout in seconds
Returns:
List of matching node information dictionaries
"""
...
def discover_commissionable_nodes_commissioning_enabled(
self,
timeout: int = 30
) -> list:
"""
Discover all nodes with commissioning enabled.
Parameters:
- timeout: Discovery timeout in seconds
Returns:
List of commissionable node information dictionaries
"""
...Utility functions for BLE operations and device management.
class ChipBleUtility:
"""Utility class for CHIP BLE operations."""
@staticmethod
def scan_devices(timeout: int = 10) -> list:
"""
Scan for CHIP-enabled BLE devices.
Parameters:
- timeout: Scan timeout in seconds
Returns:
List of device dictionaries with name, address, and RSSI
"""
...
@staticmethod
def connect_device(device_address: str) -> bool:
"""
Connect to BLE device by address.
Parameters:
- device_address: BLE device MAC address
Returns:
True if connection successful
"""
...
@staticmethod
def disconnect_device(device_address: str) -> bool:
"""
Disconnect from BLE device.
Parameters:
- device_address: BLE device MAC address
Returns:
True if disconnection successful
"""
...
@staticmethod
def get_device_info(device_address: str) -> dict:
"""
Get detailed information about a BLE device.
Parameters:
- device_address: BLE device MAC address
Returns:
Dictionary with device information or None if not found
"""
...
@staticmethod
def is_chip_device(device_address: str) -> bool:
"""
Check if BLE device supports CHIP protocol.
Parameters:
- device_address: BLE device MAC address
Returns:
True if device supports CHIP
"""
...from chip.discovery import DiscoveryManager
from chip.ChipCommissionableNodeCtrl import ChipCommissionableNodeCtrl
# Initialize discovery manager
discovery = DiscoveryManager()
# Discover all commissionable devices
devices = discovery.discover_commissionable_nodes(timeout=30)
for device in devices:
print(f"Found device: {device['instanceName']}")
print(f" Vendor ID: {device['vendorId']}")
print(f" Product ID: {device['productId']}")
print(f" Discriminator: {device['longDiscriminator']}")
print(f" IP Address: {device['addresses']}")
# Discover devices by vendor ID
commissionable = ChipCommissionableNodeCtrl()
vendor_devices = commissionable.discover_commissionable_nodes_vendor_id(
vendor_id=0x1234,
timeout=30
)
# Discover operational devices
operational_devices = discovery.discover_operational_devices(timeout=30)
print(f"Found {len(operational_devices)} operational devices")from chip.ble import BleManager
from chip.ChipBleUtility import ChipBleUtility
# Scan for BLE devices
ble_devices = ChipBleUtility.scan_devices(timeout=15)
for device in ble_devices:
print(f"BLE Device: {device['name']} ({device['address']}) RSSI: {device['rssi']}")
# Check if it's a CHIP device
if ChipBleUtility.is_chip_device(device['address']):
print(f" -> CHIP-enabled device found!")
# Get detailed info
info = ChipBleUtility.get_device_info(device['address'])
if info:
print(f" Services: {info.get('services', [])}")
# Connect to a specific BLE device
ble_manager = BleManager()
success = ble_manager.connect("AA:BB:CC:DD:EE:FF")
if success:
print("BLE connection established")
# Use the BLE connection for commissioning
# (typically done through ChipDeviceController.CommissionWithCode)
# Disconnect when done
ble_manager.disconnect()import platform
# Use platform-appropriate BLE manager
if platform.system() == "Linux":
from chip.ChipBluezMgr import ChipBluezMgr
ble_mgr = ChipBluezMgr(adapter_name="hci0")
ble_mgr.start_scan()
# Wait for scan results...
import time
time.sleep(10)
ble_mgr.stop_scan()
elif platform.system() == "Darwin": # macOS
from chip.ChipCoreBluetoothMgr import ChipCoreBluetoothMgr
ble_mgr = ChipCoreBluetoothMgr()
# Scan for CHIP service UUIDs
chip_service_uuids = ["0000FFF6-0000-1000-8000-00805F9B34FB"]
ble_mgr.start_scan(service_uuids=chip_service_uuids)
# Process scan results...
time.sleep(10)
ble_mgr.stop_scan()from chip.ChipDeviceCtrl import ChipDeviceController
from chip.discovery import DiscoveryManager
from chip.ble import BleManager
# Initialize controller and discovery
controller = ChipDeviceController(controllerNodeId=12345)
discovery = DiscoveryManager()
# First try network discovery
network_devices = discovery.discover_commissionable_nodes(timeout=30)
if network_devices:
# Commission via network
device = network_devices[0]
success = controller.CommissionOnNetwork(
nodeId=1,
setupPinCode=20202021,
filterType=2, # Long discriminator
filter=str(device['longDiscriminator'])
)
print(f"Network commissioning: {'Success' if success else 'Failed'}")
else:
# Fall back to BLE commissioning
print("No network devices found, trying BLE...")
ble_manager = BleManager()
ble_devices = ble_manager.scan(timeout=15)
if ble_devices:
# Connect to first CHIP BLE device
first_device = ble_devices[0]
if ble_manager.connect(first_device['identifier']):
# Commission via BLE (using setup payload)
setup_payload = "MT:YNJV00KA0648G00" # Example payload
success = controller.CommissionWithCode(
setupPayload=setup_payload,
nodeId=1,
discoveryType=1 # BLE only
)
print(f"BLE commissioning: {'Success' if success else 'Failed'}")
ble_manager.disconnect()Install with Tessl CLI
npx tessl i tessl/pypi-home-assistant-chip-clusters