How to communicate between CustomClusters using Bus
devWhen a quirk requires translating data from one cluster to another, use the Bus utility class.
- Initialize Buses: In the
__init__method of yourCustomDevice, create instances ofBus(e.g.,self.power_bus = Bus()). - Publish Events: In your source
CustomCluster, override_update_attributeto catch incoming data and publish it to the bus usingself.endpoint.device.bus_name.listener_event(EVENT_NAME, value). - Subscribe to Events: In your target
CustomCluster, useself.endpoint.device.bus_name.add_listener(self)in the__init__method. - Handle Events: Implement a method in the target cluster where the method name matches the
EVENT_NAMEused inlistener_eventexactly. This method will receive the value and can then callself._update_attributeto update the local cluster state.
# 1. In the Device
class MyDevice(CustomDevice):
def __init__(self, *args, **kwargs):
self.my_bus = Bus()
super().__init__(*args, **kwargs)
# 2. In the Source Cluster
class SourceCluster(CustomCluster, StandardCluster):
def _update_attribute(self, attrid, value):
super()._update_attribute(attrid, value)
if value is not None:
# Publish the event
self.endpoint.device.my_bus.listener_event(MY_EVENT, value)
# 3. In the Target Cluster
class TargetCluster(CustomCluster, StandardCluster):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Subscribe to the bus
self.endpoint.device.my_bus.add_listener(self)
def my_event(self, value):
# This method name must match MY_EVENT exactly
self._update_attribute(SOME_ATTRIBUTE_ID, value)