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
—
TLV (Tag-Length-Value) encoding and decoding for Matter protocol data structures, enabling low-level protocol operations and custom data handling.
Encode data into TLV format for Matter protocol transmission.
class TLVWriter:
"""TLV encoder for Matter protocol data structures."""
def __init__(self):
"""Initialize TLV writer."""
...
def put(self, tag: int, value) -> bool:
"""
Write a tagged value to the TLV stream.
Parameters:
- tag: TLV tag identifier
- value: Value to encode (int, float, str, bytes, bool, list, dict)
Returns:
True if write successful
"""
...
def put_null(self, tag: int) -> bool:
"""
Write a null value.
Parameters:
- tag: TLV tag identifier
Returns:
True if write successful
"""
...
def put_bool(self, tag: int, value: bool) -> bool:
"""
Write a boolean value.
Parameters:
- tag: TLV tag identifier
- value: Boolean value
Returns:
True if write successful
"""
...
def put_int(self, tag: int, value: int) -> bool:
"""
Write a signed integer value.
Parameters:
- tag: TLV tag identifier
- value: Integer value
Returns:
True if write successful
"""
...
def put_uint(self, tag: int, value: int) -> bool:
"""
Write an unsigned integer value.
Parameters:
- tag: TLV tag identifier
- value: Unsigned integer value
Returns:
True if write successful
"""
...
def put_float(self, tag: int, value: float) -> bool:
"""
Write a floating-point value.
Parameters:
- tag: TLV tag identifier
- value: Float value (32-bit or 64-bit)
Returns:
True if write successful
"""
...
def put_string(self, tag: int, value: str) -> bool:
"""
Write a UTF-8 string value.
Parameters:
- tag: TLV tag identifier
- value: String value
Returns:
True if write successful
"""
...
def put_bytes(self, tag: int, value: bytes) -> bool:
"""
Write a byte string value.
Parameters:
- tag: TLV tag identifier
- value: Byte string value
Returns:
True if write successful
"""
...
def start_container(self, tag: int, container_type: int) -> bool:
"""
Start a TLV container (structure, array, or list).
Parameters:
- tag: TLV tag identifier
- container_type: Container type (0=Structure, 1=Array, 2=List)
Returns:
True if container started successfully
"""
...
def end_container(self) -> bool:
"""
End the current TLV container.
Returns:
True if container ended successfully
"""
...
def finalize(self) -> bytes:
"""
Finalize the TLV encoding and return the encoded data.
Returns:
Encoded TLV data as bytes
"""
...
def get_length(self) -> int:
"""
Get the length of encoded data so far.
Returns:
Number of encoded bytes
"""
...
def reset(self):
"""Reset the writer to start encoding new data."""
...Decode TLV formatted data from Matter protocol messages.
class TLVReader:
"""TLV decoder for Matter protocol data structures."""
def __init__(self, data: bytes):
"""
Initialize TLV reader with encoded data.
Parameters:
- data: TLV encoded data bytes
"""
...
def get(self, tag: int):
"""
Get value by tag.
Parameters:
- tag: TLV tag identifier to find
Returns:
Decoded value or None if tag not found
"""
...
def get_bool(self, tag: int) -> bool:
"""
Get boolean value by tag.
Parameters:
- tag: TLV tag identifier
Returns:
Boolean value or None if tag not found
Raises:
ValueError: If tag exists but is not a boolean
"""
...
def get_int(self, tag: int) -> int:
"""
Get signed integer value by tag.
Parameters:
- tag: TLV tag identifier
Returns:
Integer value or None if tag not found
Raises:
ValueError: If tag exists but is not an integer
"""
...
def get_uint(self, tag: int) -> int:
"""
Get unsigned integer value by tag.
Parameters:
- tag: TLV tag identifier
Returns:
Unsigned integer value or None if tag not found
Raises:
ValueError: If tag exists but is not an unsigned integer
"""
...
def get_float(self, tag: int) -> float:
"""
Get floating-point value by tag.
Parameters:
- tag: TLV tag identifier
Returns:
Float value or None if tag not found
Raises:
ValueError: If tag exists but is not a float
"""
...
def get_string(self, tag: int) -> str:
"""
Get UTF-8 string value by tag.
Parameters:
- tag: TLV tag identifier
Returns:
String value or None if tag not found
Raises:
ValueError: If tag exists but is not a string
"""
...
def get_bytes(self, tag: int) -> bytes:
"""
Get byte string value by tag.
Parameters:
- tag: TLV tag identifier
Returns:
Byte string value or None if tag not found
Raises:
ValueError: If tag exists but is not a byte string
"""
...
def get_all(self) -> dict:
"""
Get all tag-value pairs.
Returns:
Dictionary mapping tags to decoded values
"""
...
def get_tags(self) -> list:
"""
Get all tags in the TLV data.
Returns:
List of tag identifiers
"""
...
def has_tag(self, tag: int) -> bool:
"""
Check if a tag exists in the TLV data.
Parameters:
- tag: TLV tag identifier
Returns:
True if tag exists
"""
...
def get_container(self, tag: int) -> 'TLVReader':
"""
Get a container (structure, array, or list) as a new TLV reader.
Parameters:
- tag: TLV tag identifier for the container
Returns:
New TLVReader for the container contents or None if not found
Raises:
ValueError: If tag exists but is not a container
"""
...
def is_container(self, tag: int) -> bool:
"""
Check if a tag contains a container.
Parameters:
- tag: TLV tag identifier
Returns:
True if tag contains a container
"""
...
def get_container_type(self, tag: int) -> int:
"""
Get the type of a container.
Parameters:
- tag: TLV tag identifier for the container
Returns:
Container type (0=Structure, 1=Array, 2=List) or None if not a container
"""
...Utility functions for common TLV operations and data type handling.
class TLVUtils:
"""Utility functions for TLV operations."""
@staticmethod
def encode_tag(profile_id: int, tag_num: int, is_context_specific: bool = False) -> int:
"""
Encode a TLV tag from profile ID and tag number.
Parameters:
- profile_id: Profile identifier (0 for anonymous)
- tag_num: Tag number within the profile
- is_context_specific: Whether this is a context-specific tag
Returns:
Encoded tag value
"""
...
@staticmethod
def decode_tag(tag: int) -> dict:
"""
Decode a TLV tag into its components.
Parameters:
- tag: Encoded tag value
Returns:
Dictionary with 'profile_id', 'tag_num', and 'is_context_specific'
"""
...
@staticmethod
def python_to_tlv(data) -> bytes:
"""
Convert Python data structure to TLV bytes.
Parameters:
- data: Python data (dict, list, primitive types)
Returns:
TLV encoded bytes
"""
...
@staticmethod
def tlv_to_python(data: bytes) -> dict:
"""
Convert TLV bytes to Python data structure.
Parameters:
- data: TLV encoded bytes
Returns:
Python data structure (dict with tag keys)
"""
...
@staticmethod
def validate_tlv(data: bytes) -> bool:
"""
Validate TLV data structure.
Parameters:
- data: TLV encoded bytes to validate
Returns:
True if TLV data is well-formed
"""
...
@staticmethod
def pretty_print_tlv(data: bytes) -> str:
"""
Create a human-readable representation of TLV data.
Parameters:
- data: TLV encoded bytes
Returns:
Formatted string representation
"""
...
@staticmethod
def get_tlv_size(data: bytes) -> int:
"""
Get the total size of TLV data.
Parameters:
- data: TLV encoded bytes
Returns:
Size in bytes
"""
...TLV handling for Matter-specific data types and structures.
class float32:
"""32-bit floating point TLV type."""
def __init__(self, value: float):
"""
Initialize 32-bit float.
Parameters:
- value: Float value
"""
...
@property
def value(self) -> float:
"""Get the float value."""
...
class uint:
"""Unsigned integer TLV type with size specification."""
def __init__(self, value: int, size: int = None):
"""
Initialize unsigned integer.
Parameters:
- value: Integer value
- size: Size in bytes (1, 2, 4, or 8), auto-detected if None
"""
...
@property
def value(self) -> int:
"""Get the integer value."""
...
@property
def size(self) -> int:
"""Get the size in bytes."""
...
class int:
"""Signed integer TLV type with size specification."""
def __init__(self, value: int, size: int = None):
"""
Initialize signed integer.
Parameters:
- value: Integer value
- size: Size in bytes (1, 2, 4, or 8), auto-detected if None
"""
...
@property
def value(self) -> int:
"""Get the integer value."""
...
@property
def size(self) -> int:
"""Get the size in bytes."""
...from chip.tlv import TLVWriter, TLVReader
# Create a TLV writer
writer = TLVWriter()
# Encode various data types
writer.put_int(1, 42) # Tag 1: integer 42
writer.put_string(2, "Hello, Matter!") # Tag 2: string
writer.put_bool(3, True) # Tag 3: boolean true
writer.put_float(4, 3.14159) # Tag 4: float
writer.put_bytes(5, b"binary_data") # Tag 5: byte string
# Get the encoded TLV data
tlv_data = writer.finalize()
print(f"Encoded TLV data: {len(tlv_data)} bytes")
# Create a TLV reader to decode the data
reader = TLVReader(tlv_data)
# Read values by tag
int_value = reader.get_int(1)
string_value = reader.get_string(2)
bool_value = reader.get_bool(3)
float_value = reader.get_float(4)
bytes_value = reader.get_bytes(5)
print(f"Integer: {int_value}")
print(f"String: {string_value}")
print(f"Boolean: {bool_value}")
print(f"Float: {float_value}")
print(f"Bytes: {bytes_value}")
# Get all values at once
all_values = reader.get_all()
print(f"All values: {all_values}")from chip.tlv import TLVWriter, TLVReader
# Create writer for structured data
writer = TLVWriter()
# Start a structure container
writer.start_container(1, 0) # Tag 1, type 0 (Structure)
# Add fields to the structure
writer.put_string(1, "Device Name") # Device name
writer.put_int(2, 0x1234) # Vendor ID
writer.put_int(3, 0x5678) # Product ID
# Start an array of endpoints
writer.start_container(4, 1) # Tag 4, type 1 (Array)
writer.put_int(0, 0) # Endpoint 0
writer.put_int(1, 1) # Endpoint 1
writer.put_int(2, 2) # Endpoint 2
writer.end_container() # End endpoints array
# End the main structure
writer.end_container()
# Get encoded data
tlv_data = writer.finalize()
print(f"Structured TLV data: {len(tlv_data)} bytes")
# Read the structured data
reader = TLVReader(tlv_data)
# Get the main structure
main_struct = reader.get_container(1)
if main_struct:
device_name = main_struct.get_string(1)
vendor_id = main_struct.get_int(2)
product_id = main_struct.get_int(3)
print(f"Device: {device_name}")
print(f"Vendor ID: 0x{vendor_id:04X}")
print(f"Product ID: 0x{product_id:04X}")
# Get the endpoints array
endpoints_array = main_struct.get_container(4)
if endpoints_array:
all_endpoints = endpoints_array.get_all()
endpoints = [all_endpoints[tag] for tag in sorted(all_endpoints.keys())]
print(f"Endpoints: {endpoints}")from chip.tlv import TLVWriter, TLVReader
import chip.clusters as Clusters
# Encode OnOff cluster command
writer = TLVWriter()
# OnOff On command (no parameters)
writer.start_container(1, 0) # Command structure
writer.end_container()
on_command_tlv = writer.finalize()
# Encode LevelControl MoveToLevel command with parameters
writer = TLVWriter()
writer.start_container(1, 0) # Command structure
writer.put_int(0, 128) # Level (0-254)
writer.put_int(1, 10) # Transition time (1/10 seconds)
writer.put_int(2, 0) # Options mask
writer.put_int(3, 0) # Options override
writer.end_container()
level_command_tlv = writer.finalize()
print(f"OnOff command TLV: {len(on_command_tlv)} bytes")
print(f"LevelControl command TLV: {len(level_command_tlv)} bytes")
# Decode level control command
reader = TLVReader(level_command_tlv)
command_struct = reader.get_container(1)
if command_struct:
level = command_struct.get_int(0)
transition_time = command_struct.get_int(1)
options_mask = command_struct.get_int(2)
options_override = command_struct.get_int(3)
print(f"Decoded MoveToLevel command:")
print(f" Level: {level}")
print(f" Transition time: {transition_time/10.0}s")
print(f" Options mask: {options_mask}")
print(f" Options override: {options_override}")from chip.tlv import TLVUtils, float32, uint, int
# Convert Python data to TLV
python_data = {
"device_info": {
"name": "Smart Light",
"endpoints": [0, 1],
"active": True,
"temperature": 23.5
},
"settings": {
"brightness": 75,
"color_temp": 2700
}
}
# Convert to TLV
tlv_bytes = TLVUtils.python_to_tlv(python_data)
print(f"Python to TLV: {len(tlv_bytes)} bytes")
# Convert back to Python
decoded_data = TLVUtils.tlv_to_python(tlv_bytes)
print(f"TLV to Python: {decoded_data}")
# Validate TLV data
is_valid = TLVUtils.validate_tlv(tlv_bytes)
print(f"TLV is valid: {is_valid}")
# Pretty print TLV structure
pretty_output = TLVUtils.pretty_print_tlv(tlv_bytes)
print("TLV Structure:")
print(pretty_output)
# Use specific TLV types
writer = TLVWriter()
# Use 32-bit float specifically
writer.put(1, float32(3.14159))
# Use unsigned integer with specific size
writer.put(2, uint(65535, size=2)) # 16-bit unsigned int
# Use signed integer with specific size
writer.put(3, int(-32768, size=2)) # 16-bit signed int
typed_tlv_data = writer.finalize()
# Read with type information preserved
reader = TLVReader(typed_tlv_data)
float_val = reader.get_float(1)
uint_val = reader.get_uint(2)
int_val = reader.get_int(3)
print(f"32-bit float: {float_val}")
print(f"16-bit uint: {uint_val}")
print(f"16-bit int: {int_val}")from chip.tlv import TLVWriter, TLVReader, TLVUtils
# Create custom tags using profile IDs
writer = TLVWriter()
# Standard Matter tags (profile 0)
standard_tag = TLVUtils.encode_tag(0, 1) # Profile 0, tag 1
writer.put_string(standard_tag, "Standard tag")
# Vendor-specific tags (custom profile)
vendor_profile = 0x1234
vendor_tag = TLVUtils.encode_tag(vendor_profile, 100)
writer.put_int(vendor_tag, 42)
# Context-specific tags
context_tag = TLVUtils.encode_tag(0, 1, is_context_specific=True)
writer.put_bool(context_tag, True)
custom_tlv_data = writer.finalize()
# Read and decode tag information
reader = TLVReader(custom_tlv_data)
all_data = reader.get_all()
for tag, value in all_data.items():
tag_info = TLVUtils.decode_tag(tag)
print(f"Tag {tag}: Profile {tag_info['profile_id']}, "
f"Number {tag_info['tag_num']}, "
f"Context-specific: {tag_info['is_context_specific']}, "
f"Value: {value}")from chip.tlv import TLVWriter, TLVReader
from chip.ChipDeviceCtrl import ChipDeviceController
import chip.clusters as Clusters
# Initialize controller
controller = ChipDeviceController(controllerNodeId=12345)
# Example: Create custom attribute write using TLV
writer = TLVWriter()
# Encode attribute write request
writer.start_container(1, 0) # AttributeWriteRequest structure
# Attribute path
writer.start_container(1, 0) # AttributePath
writer.put_int(1, 1) # Endpoint
writer.put_int(2, 6) # Cluster ID (OnOff)
writer.put_int(3, 0) # Attribute ID (OnOff)
writer.end_container()
# Attribute value
writer.put_bool(2, True) # Turn on
writer.end_container()
write_request_tlv = writer.finalize()
# In practice, this TLV would be sent as part of a Matter message
print(f"Custom attribute write TLV: {len(write_request_tlv)} bytes")
# Example: Parse attribute read response TLV
# (This would typically come from a device response)
response_tlv = b"..." # Simulated response data
try:
reader = TLVReader(response_tlv)
# Parse attribute read response structure
if reader.has_tag(1): # Success response
attr_data = reader.get_container(1)
if attr_data:
# Extract attribute path and value
path = attr_data.get_container(1)
value = attr_data.get(2)
if path:
endpoint = path.get_int(1)
cluster = path.get_int(2)
attribute = path.get_int(3)
print(f"Attribute read response:")
print(f" Endpoint: {endpoint}")
print(f" Cluster: {cluster}")
print(f" Attribute: {attribute}")
print(f" Value: {value}")
elif reader.has_tag(2): # Error response
error_code = reader.get_int(2)
print(f"Attribute read failed with error: {error_code}")
except Exception as e:
print(f"Failed to parse TLV response: {e}")
# Clean up
controller.Shutdown()Install with Tessl CLI
npx tessl i tessl/pypi-home-assistant-chip-clusters