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
storage.mddocs/
Storage
Storage interfaces for device credentials, fabric information, and configuration persistence across application restarts.
Capabilities
Persistent Storage Interface
Core persistent storage interface for Matter stack data.
class PersistentStorage:
"""Persistent storage interface for CHIP stack data."""
def __init__(self, path: str):
"""
Initialize persistent storage.
Parameters:
- path: File system path for storage
"""
...
def set(self, key: str, value: bytes) -> bool:
"""
Store a key-value pair.
Parameters:
- key: Storage key string
- value: Binary data to store
Returns:
True if storage successful
"""
...
def get(self, key: str) -> bytes:
"""
Retrieve value by key.
Parameters:
- key: Storage key string
Returns:
Binary data or None if key not found
"""
...
def delete(self, key: str) -> bool:
"""
Delete a key-value pair.
Parameters:
- key: Storage key to delete
Returns:
True if deletion successful
"""
...
def clear(self) -> bool:
"""
Clear all stored data.
Returns:
True if clear successful
"""
...
def keys(self) -> list:
"""
Get all storage keys.
Returns:
List of all keys in storage
"""
...
def exists(self, key: str) -> bool:
"""
Check if key exists in storage.
Parameters:
- key: Storage key to check
Returns:
True if key exists
"""
...
def size(self) -> int:
"""
Get total number of stored items.
Returns:
Number of key-value pairs in storage
"""
...
def sync(self) -> bool:
"""
Force synchronization to persistent media.
Returns:
True if sync successful
"""
...Configuration Storage
High-level interface for storing configuration data.
class ConfigurationStorage:
"""High-level configuration storage interface."""
def __init__(self, storage: PersistentStorage):
"""
Initialize configuration storage.
Parameters:
- storage: Underlying persistent storage instance
"""
...
def set_device_config(self, node_id: int, config: dict) -> bool:
"""
Store device-specific configuration.
Parameters:
- node_id: Device node ID
- config: Configuration dictionary
Returns:
True if storage successful
"""
...
def get_device_config(self, node_id: int) -> dict:
"""
Retrieve device configuration.
Parameters:
- node_id: Device node ID
Returns:
Configuration dictionary or None if not found
"""
...
def delete_device_config(self, node_id: int) -> bool:
"""
Delete device configuration.
Parameters:
- node_id: Device node ID
Returns:
True if deletion successful
"""
...
def set_fabric_config(self, fabric_id: int, config: dict) -> bool:
"""
Store fabric-specific configuration.
Parameters:
- fabric_id: Fabric ID
- config: Fabric configuration dictionary
Returns:
True if storage successful
"""
...
def get_fabric_config(self, fabric_id: int) -> dict:
"""
Retrieve fabric configuration.
Parameters:
- fabric_id: Fabric ID
Returns:
Fabric configuration dictionary or None if not found
"""
...
def set_controller_config(self, controller_id: int, config: dict) -> bool:
"""
Store controller configuration.
Parameters:
- controller_id: Controller node ID
- config: Controller configuration dictionary
Returns:
True if storage successful
"""
...
def get_controller_config(self, controller_id: int) -> dict:
"""
Retrieve controller configuration.
Parameters:
- controller_id: Controller node ID
Returns:
Controller configuration dictionary or None if not found
"""
...
def list_devices(self) -> list:
"""
List all devices with stored configurations.
Returns:
List of device node IDs
"""
...
def list_fabrics(self) -> list:
"""
List all fabrics with stored configurations.
Returns:
List of fabric IDs
"""
...Credential Storage
Secure storage for certificates, keys, and other security credentials.
class CredentialStorage:
"""Secure storage for security credentials."""
def __init__(self, storage: PersistentStorage):
"""
Initialize credential storage.
Parameters:
- storage: Underlying persistent storage instance
"""
...
def store_fabric_credentials(
self,
fabric_id: int,
root_cert: bytes,
intermediate_cert: bytes = None,
operational_cert: bytes = None,
private_key: bytes = None
) -> bool:
"""
Store fabric credentials.
Parameters:
- fabric_id: Fabric ID
- root_cert: Root certificate (required)
- intermediate_cert: Intermediate certificate (optional)
- operational_cert: Operational certificate (optional)
- private_key: Private key (optional)
Returns:
True if storage successful
"""
...
def get_fabric_credentials(self, fabric_id: int) -> dict:
"""
Retrieve fabric credentials.
Parameters:
- fabric_id: Fabric ID
Returns:
Dictionary with credential data or None if not found
"""
...
def store_device_credentials(
self,
node_id: int,
fabric_id: int,
device_cert: bytes,
device_key: bytes = None
) -> bool:
"""
Store device-specific credentials.
Parameters:
- node_id: Device node ID
- fabric_id: Associated fabric ID
- device_cert: Device certificate
- device_key: Device private key (optional)
Returns:
True if storage successful
"""
...
def get_device_credentials(self, node_id: int, fabric_id: int) -> dict:
"""
Retrieve device credentials.
Parameters:
- node_id: Device node ID
- fabric_id: Fabric ID
Returns:
Dictionary with credential data or None if not found
"""
...
def store_paa_certificates(self, paa_certs: list) -> bool:
"""
Store Product Attestation Authority certificates.
Parameters:
- paa_certs: List of PAA certificate bytes
Returns:
True if storage successful
"""
...
def get_paa_certificates(self) -> list:
"""
Retrieve all PAA certificates.
Returns:
List of PAA certificate bytes
"""
...
def delete_fabric_credentials(self, fabric_id: int) -> bool:
"""
Delete all credentials for a fabric.
Parameters:
- fabric_id: Fabric ID
Returns:
True if deletion successful
"""
...
def delete_device_credentials(self, node_id: int, fabric_id: int) -> bool:
"""
Delete device credentials.
Parameters:
- node_id: Device node ID
- fabric_id: Fabric ID
Returns:
True if deletion successful
"""
...Session Storage
Storage for session information and security contexts.
class SessionStorage:
"""Storage for session information."""
def __init__(self, storage: PersistentStorage):
"""
Initialize session storage.
Parameters:
- storage: Underlying persistent storage instance
"""
...
def store_session(
self,
node_id: int,
fabric_id: int,
session_data: dict
) -> bool:
"""
Store session information.
Parameters:
- node_id: Device node ID
- fabric_id: Fabric ID
- session_data: Session information dictionary
Returns:
True if storage successful
"""
...
def get_session(self, node_id: int, fabric_id: int) -> dict:
"""
Retrieve session information.
Parameters:
- node_id: Device node ID
- fabric_id: Fabric ID
Returns:
Session information dictionary or None if not found
"""
...
def delete_session(self, node_id: int, fabric_id: int) -> bool:
"""
Delete session information.
Parameters:
- node_id: Device node ID
- fabric_id: Fabric ID
Returns:
True if deletion successful
"""
...
def cleanup_expired_sessions(self) -> int:
"""
Remove expired session information.
Returns:
Number of sessions cleaned up
"""
...
def get_active_sessions(self) -> list:
"""
Get list of active sessions.
Returns:
List of (node_id, fabric_id) tuples
"""
...Subscription Storage
Persistent storage for subscription information.
class SubscriptionStorage:
"""Storage for subscription information."""
def __init__(self, storage: PersistentStorage):
"""
Initialize subscription storage.
Parameters:
- storage: Underlying persistent storage instance
"""
...
def store_subscription(
self,
subscription_id: int,
node_id: int,
fabric_id: int,
subscription_data: dict
) -> bool:
"""
Store subscription information.
Parameters:
- subscription_id: Unique subscription ID
- node_id: Target device node ID
- fabric_id: Fabric ID
- subscription_data: Subscription configuration dictionary
Returns:
True if storage successful
"""
...
def get_subscription(self, subscription_id: int) -> dict:
"""
Retrieve subscription information.
Parameters:
- subscription_id: Subscription ID
Returns:
Subscription information dictionary or None if not found
"""
...
def get_subscriptions_for_device(self, node_id: int, fabric_id: int) -> list:
"""
Get all subscriptions for a specific device.
Parameters:
- node_id: Device node ID
- fabric_id: Fabric ID
Returns:
List of subscription information dictionaries
"""
...
def delete_subscription(self, subscription_id: int) -> bool:
"""
Delete subscription information.
Parameters:
- subscription_id: Subscription ID
Returns:
True if deletion successful
"""
...
def cleanup_orphaned_subscriptions(self) -> int:
"""
Remove subscriptions for non-existent devices.
Returns:
Number of subscriptions cleaned up
"""
...Usage Examples
Basic Persistent Storage
from chip.storage import PersistentStorage
# Initialize storage
storage = PersistentStorage("/var/lib/chip/storage")
# Store configuration data
config_data = b'{"setting1": "value1", "setting2": 42}'
success = storage.set("app_config", config_data)
print(f"Storage success: {success}")
# Retrieve data
retrieved_data = storage.get("app_config")
if retrieved_data:
print(f"Retrieved: {retrieved_data}")
# Check if key exists
if storage.exists("app_config"):
print("Configuration found")
# List all keys
all_keys = storage.keys()
print(f"All stored keys: {all_keys}")
# Get storage size
print(f"Total items: {storage.size()}")
# Force sync to disk
storage.sync()
# Clean up
storage.delete("app_config")Configuration Storage
from chip.storage import PersistentStorage, ConfigurationStorage
# Initialize storage layers
storage = PersistentStorage("/var/lib/chip/config")
config_storage = ConfigurationStorage(storage)
# Store device configuration
device_config = {
"name": "Living Room Light",
"type": "dimmable_light",
"endpoints": [1],
"clusters": ["OnOff", "LevelControl", "ColorControl"],
"last_seen": "2024-09-10T13:30:00Z"
}
success = config_storage.set_device_config(node_id=1, config=device_config)
print(f"Device config stored: {success}")
# Store fabric configuration
fabric_config = {
"name": "Home Fabric",
"created": "2024-09-01T10:00:00Z",
"max_devices": 100,
"security_policy": "strict"
}
config_storage.set_fabric_config(fabric_id=1, config=fabric_config)
# Retrieve configurations
device_info = config_storage.get_device_config(node_id=1)
if device_info:
print(f"Device name: {device_info['name']}")
print(f"Device type: {device_info['type']}")
fabric_info = config_storage.get_fabric_config(fabric_id=1)
if fabric_info:
print(f"Fabric name: {fabric_info['name']}")
# List all devices and fabrics
devices = config_storage.list_devices()
fabrics = config_storage.list_fabrics()
print(f"Configured devices: {devices}")
print(f"Configured fabrics: {fabrics}")Credential Storage
from chip.storage import PersistentStorage, CredentialStorage
# Initialize credential storage
storage = PersistentStorage("/var/lib/chip/credentials")
cred_storage = CredentialStorage(storage)
# Store fabric credentials (example with dummy data)
root_cert = b"-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"
operational_cert = b"-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"
private_key = b"-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"
success = cred_storage.store_fabric_credentials(
fabric_id=1,
root_cert=root_cert,
operational_cert=operational_cert,
private_key=private_key
)
print(f"Fabric credentials stored: {success}")
# Store PAA certificates
paa_certs = [
b"-----BEGIN CERTIFICATE-----\nPAA1...\n-----END CERTIFICATE-----",
b"-----BEGIN CERTIFICATE-----\nPAA2...\n-----END CERTIFICATE-----"
]
cred_storage.store_paa_certificates(paa_certs)
# Retrieve credentials
fabric_creds = cred_storage.get_fabric_credentials(fabric_id=1)
if fabric_creds:
print("Fabric credentials found:")
print(f" Root cert length: {len(fabric_creds.get('root_cert', b''))}")
print(f" Operational cert available: {'operational_cert' in fabric_creds}")
print(f" Private key available: {'private_key' in fabric_creds}")
# Get PAA certificates
paa_list = cred_storage.get_paa_certificates()
print(f"Number of PAA certificates: {len(paa_list)}")Session and Subscription Storage
from chip.storage import PersistentStorage, SessionStorage, SubscriptionStorage
import time
# Initialize storage
storage = PersistentStorage("/var/lib/chip/sessions")
session_storage = SessionStorage(storage)
subscription_storage = SubscriptionStorage(storage)
# Store session information
session_data = {
"session_id": "abc123",
"established": time.time(),
"last_activity": time.time(),
"encryption_key": "key_data_here",
"expires": time.time() + 3600 # 1 hour from now
}
session_storage.store_session(
node_id=1,
fabric_id=1,
session_data=session_data
)
# Store subscription information
subscription_data = {
"attributes": [
{"endpoint": 1, "cluster": "OnOff", "attribute": "OnOff"},
{"endpoint": 1, "cluster": "LevelControl", "attribute": "CurrentLevel"}
],
"min_interval": 1,
"max_interval": 10,
"created": time.time(),
"active": True
}
subscription_storage.store_subscription(
subscription_id=12345,
node_id=1,
fabric_id=1,
subscription_data=subscription_data
)
# Retrieve session
session_info = session_storage.get_session(node_id=1, fabric_id=1)
if session_info:
print(f"Session established: {session_info['established']}")
print(f"Session expires: {session_info['expires']}")
# Get subscriptions for device
device_subscriptions = subscription_storage.get_subscriptions_for_device(
node_id=1,
fabric_id=1
)
print(f"Device has {len(device_subscriptions)} subscriptions")
# Get active sessions
active_sessions = session_storage.get_active_sessions()
print(f"Active sessions: {active_sessions}")
# Cleanup expired sessions
cleaned_sessions = session_storage.cleanup_expired_sessions()
print(f"Cleaned up {cleaned_sessions} expired sessions")Integrated Storage Management
from chip.storage import PersistentStorage, ConfigurationStorage, CredentialStorage
from chip.ChipDeviceCtrl import ChipDeviceController
# Initialize comprehensive storage system
base_storage = PersistentStorage("/var/lib/chip")
config_storage = ConfigurationStorage(base_storage)
cred_storage = CredentialStorage(base_storage)
# Initialize controller with storage
controller = ChipDeviceController(
controllerNodeId=12345,
persistentStoragePath="/var/lib/chip"
)
try:
# Commission a device and store its configuration
success = controller.CommissionOnNetwork(
nodeId=1,
setupPinCode=20202021
)
if success:
# Wait for commissioning to complete...
# Then store device configuration
device_config = {
"commissioned": time.time(),
"vendor_id": 0x1234,
"product_id": 0x5678,
"device_type": "light",
"endpoints": [1],
"clusters": ["OnOff", "LevelControl"]
}
config_storage.set_device_config(node_id=1, config=device_config)
# Store fabric information
fabric_config = {
"fabric_label": "My Smart Home",
"created": time.time(),
"devices_count": 1
}
config_storage.set_fabric_config(fabric_id=1, config=fabric_config)
print("Device commissioned and configuration stored")
# Later, retrieve device information
stored_config = config_storage.get_device_config(node_id=1)
if stored_config:
print(f"Device type: {stored_config['device_type']}")
print(f"Commissioned on: {stored_config['commissioned']}")
finally:
controller.Shutdown()Install with Tessl CLI
npx tessl i tessl/pypi-home-assistant-chip-clusters