docs/development_controllers/matter-repl/Matter_Basic_Interactions.ipynb
This walks through the various interactions that can be initiated from the REPL towards a target using the Matter Interaction Model (IM) and Data Model (DM).
Let's clear out our persisted storage (if one exists) to start from a clean slate.
import os, subprocess
if os.path.isfile('/tmp/repl-storage.json'):
os.remove('/tmp/repl-storage.json')
# So that the all-clusters-app won't boot with stale prior state.
os.system('rm -rf /tmp/chip_*')
Let's first begin by setting up by importing some key modules that are needed to make it easier for us to interact with the Matter stack.
ReplStartup.py is run within the global namespace. This results in all of its imports being made available here.
NOTE: This is not needed if you launch the REPL from the command-line.
%reset -f
import importlib.util
spec = importlib.util.find_spec('matter.ReplStartup')
%run {spec.origin}
The Interaction Model uses data model types that refer not just to the various base types defines in the spec, but types that correspond to structs/commands/events/attributes defined in each cluster specification. The cluster-specific types are referred to as 'cluster objects'. These are represented as Python dataclasses, with each field in the respective object equivalently named as a member within the dataclass.
Objects in clusters are organized into namespaces. All clusters can be found under the Clusters namespace, with the appropriate cluster in upper camel case within that. (e.g Clusters.UnitTesting).
Within that, Commands, Structs and Attributes delimit the respective types in the cluster.
Example Struct:
v = Clusters.UnitTesting.Structs.SimpleStruct()
v.a = 20
v.b = True
v.c = Clusters.UnitTesting.Enums.SimpleEnum.kValueA
v.d = b'1234'
v.e = 30
v.g = 23.234
v
Example Command:
Clusters.UnitTesting.Commands.TestAddArguments()
To get more information about the fields in these objects and their types, run:
matterhelp(Clusters.UnitTesting.Commands.TestAddArguments)
For fields that are nullable, they are represented as a Typing.Union[Nullable, ...]. This means that it can either be a Nullable type or the underlying type of the field.
When nullable, a field can either take on the value of the native type, or a value of NullValue.
a = Clusters.UnitTesting.Structs.NullablesAndOptionalsStruct()
a.nullableInt = Clusters.Types.NullValue
a
If a field is optional, it is represented in the typing hints as a Typing.Union[NoneType, ...]. An optional field that isn't present has a value of None.
print(a.nullableOptionalInt)
Upon construction of a cluster object, the fields within that object are automatically initialized to type specific defaults as specified in the data model specification:
Clusters.UnitTesting.Structs.SimpleStruct()
This section will walk through the various types of IM Interactions that are possible in the REPL.
Let's launch an instance of the chip-all-clusters-app.
NOTE: If you're interacting with real devices, this step can be skipped.
import time, os
import subprocess
os.system('pkill -f chip-all-clusters-app')
time.sleep(1)
appPath = '../../../out/linux-x64-all-clusters/chip-all-clusters-app'
process = subprocess.Popen(appPath, stdout=subprocess.DEVNULL)
time.sleep(1)
devices = await devCtrl.DiscoverCommissionableNodes(filterType=matter.discovery.FilterType.LONG_DISCRIMINATOR, filter=3840, stopOnFirst=True, timeoutSecond=2)
devices
You can find a list of discovered device
You can call Commission(nodeId, setupPinCode) on one of the returned object:
await devices[0].Commission(2, 20202021)
The device will be commissioned by the DeviceController instance that discovered it (caIndex(1)/fabricId(0x0000000000000001)/nodeId(0x000000000001B669) in this case).
Commission the target with a NodeId of 2.
WARNING: The device can only be commissioned once. Repeating the commissioning process will result in Errors. The line
%%script truehas been added to bypass execution errors and allow the notebook to run automatically.
If you wish to test the behaviour of the function, remove %%script true
%%script true
await devCtrl.CommissionOnNetwork(2, 20202021)
To commission a Thread-based target over BLE, ensure your BLE stack is up on your host and available as hci0 on Linux. You can confirm this by running hciconfig -a. You'll also need Thread credentials to join the Thread network.
NOTE: MacOS Monterey is currently not supported due to issues with its BLE stack.
%%script true
await devCtrl.CommissionThread(3840, 20202021, 2, b'\x01\x03\xff')
To commission a Wifi-based target over BLE, ensure your BLE stack is up on your host and available as hci0 on Linux. You can confirm this by running hciconfig -a. You'll also need Wifi credentials to join the WiFi network.
NOTE: MacOS Monterey is currently not supported due to issues with its BLE stack.
%%script true
await devCtrl.CommissionWiFi(3840, 20202021, 2, 'MyWifiSsid', 'MyWifiPassword')
Let's send a basic command to turn on/off the light on Endpoint 1.
await devCtrl.SendCommand(2, 1, Clusters.OnOff.Commands.On())
The receipt of a successful status response will result in the command just returning successfully. Otherwise, an exception will be thrown.
If we send the same command to an invalid endpoint, an exception is thrown. If an IM status code was received from the server, a InteractionModelError is thrown containing the IM status code:
from pprint import pprint
try:
await devCtrl.SendCommand(2, 100, Clusters.OnOff.Commands.On())
except Exception as e:
pprint(e)
Here's an example of a command that sends back a data response, and how that is presented:
await devCtrl.SendCommand(2, 1, Clusters.UnitTesting.Commands.TestListInt8UReverseRequest([1, 3, 5, 7]))
The ReadAttribute method on the DeviceController class can be used to read attributes from a target. The NodeId of the target is the first argument, followed by a list of paths that are expressed as cluster object namespaces to the respective slices of the data that is requested.
By default, the data is returned as a dictionary, with the top-level item representing the endpoint, then the cluster and the attribute. The latter two keys are expressed using cluster object namespaces.
a = await devCtrl.ReadAttribute(2, [Clusters.UnitTesting.Attributes.Int16u])
a
a[1][Clusters.UnitTesting]
a[1][Clusters.UnitTesting][Clusters.UnitTesting.Attributes.Int16u]
await devCtrl.ReadAttribute(2, [Clusters.UnitTesting.Attributes.Int16u, Clusters.UnitTesting.Attributes.Boolean])
The path is represented as tuple of (endpoint, cluster)
await devCtrl.ReadAttribute(2, [(1, Clusters.OnOff)])
await devCtrl.ReadAttribute(2, [Clusters.OnOff])
await devCtrl.ReadAttribute(2, [2])
await devCtrl.ReadAttribute(2, [('*')])
The above encapsulates each attribute as a 'cluster-object' key within the top-level cluster instance. Instead, an alternative view each attribute is represented as a field in the object can be retrieved by passing in True to the argument returnClusterObject:
await devCtrl.ReadAttribute(2, [2], returnClusterObject=True)
A ReadEvent API exists that behaves similarly to the ReadAttribute API. It permits the same degrees of wildcard expression as its counterpart and follows the same format for expressing all wildcard permutations.
# Force an event to get emitted.
await devCtrl.SendCommand(2, 1, Clusters.UnitTesting.Commands.TestEmitTestEventRequest())
await devCtrl.ReadEvent(2, [('*')])
To subscribe to a Node, the same ReadAttribute API is used to trigger a subscription, with a valid reportInterval tuple passed in being used as a way to indicate the request to create a subscription.
reportingTimingParams = (0, 2) # MinInterval = 0s, MaxInterval = 2s
subscription = await devCtrl.ReadAttribute(2, [(2, Clusters.OnOff)], returnClusterObject=True, reportInterval=reportingTimingParams)
subscription
subscription.GetAttributes()
To trigger a report, let's alter the state of the on/off switch on EP1. That should trigger the generation of a set of attribute reports.
The SubscriptionTransaction object returned by ReadAttribute permits installing a callback that is invoked on any attribute report. A default callback is installed above that just dumps out the attribute data.
await devCtrl.SendCommand(2, 2, Clusters.OnOff.Commands.On())
time.sleep(1)
await devCtrl.SendCommand(2, 2, Clusters.OnOff.Commands.Off())
time.sleep(1)
subscription.Shutdown()
reportingTimingParams = (0, 2) # MinInterval = 0s, MaxInterval = 2s
# Subscribing to Events from EndPoint 1
subscription = await devCtrl.ReadEvent(2, [ 1 ], reportInterval = reportingTimingParams)
subscription.GetEvents()
Force an event to get emitted, which after a short while, should generate a report and trigger the print out of the received event:
await devCtrl.SendCommand(2, 1, Clusters.UnitTesting.Commands.TestEmitTestEventRequest())
time.sleep(3)
subscription.Shutdown()
To write attribute data, the WriteAttribute API can be used. It requires a NodeId and a list of cluster object encapsulated data for the attribute being written.
await devCtrl.WriteAttribute(2, [ (1, Clusters.UnitTesting.Attributes.Int16u(2)) ])
await devCtrl.ReadAttribute(2, [ (1, Clusters.UnitTesting.Attributes.Int16u) ])