How to parse/access an incoming ZCL frame?

[garrethcain]

Hopefully this is the right place, if not please direct me.

I'm trying to implement a super lightweight Zigbee implementation to send and receive messages from an MQTT broker and have chosen Zigpy and Zigpy-Znp.

I've reached the stage where I need to parse the message and publish it to my MQTT broker, I can see the following in the logs;

zigpy.zcl[33287] DEBUG [0x5EA2:1:0x0006] Received ZCL frame: b'\x11,\x01'
zigpy.zcl[33287] DEBUG [0x5EA2:1:0x0006] Decoded ZCL frame header: ZCLHeader(frame_control=FrameControl(frame_type=<FrameType.CLUSTER_COMMAND: 1>, is_manufacturer_specific=0, direction=<Direction.Server_to_Client: 0>, disable_default_response=1, reserved=0, *is_cluster=True, *is_general=False), tsn=44, command_id=1, *direction=<Direction.Server_to_Client: 0>)
zigpy.zcl[33287] DEBUG [0x5EA2:1:0x0006] Decoded ZCL frame: OnOff:on()
zigpy.zcl[33287] DEBUG [0x5EA2:1:0x0006] Received command 0x01 (TSN 44): on()
zigpy.zcl[33287] DEBUG [0x5EA2:1:0x0006] No explicit handler for cluster command 0x01: on()

Essentially I want access to the "Decoded ZCL frame" so I can pass it on to where it needs to go as it seem to have what I want. I have implemented my own handle_message method, which is receiving data.

I just want to know how to access the "decoded" ZCL frame or how to, as is implied in the log above, implement my own explicit handler for the cluster command.

Any help super appreciated and many thanks in advance.

[garrethcain]

Even just a link to a how-to would be appreciated.

[puddly]

To access incoming ZCL frames, you would have to attach a listener to the cluster object on the device you're interested in receiving frames:

zigpy/zigpy/zcl/__init__.py

Line 427 in e54dab0

self.listener_event("cluster_command", hdr.tsn, hdr.command_id, args)

Something like this:

class MyListener:
    def __init__(self, endpoint):
        self.endpoint = endpoint
        self.device = endpoint.device

    def cluster_command(self, tsn, command_id, cmd):
        print(f"Device {self.device} received command {cmd} on endpoint {self.endpoint}")

ep = dev.endpoints[1]
ep.on_off.add_listener(MyListener(ep))

Alternatively, you can create a listener for the device:

def my_callback(cmd_hdr, cmd):
    print("Received command {cmd_hdr} {cmd}")

with app._callback_for_response(src=dev, filters=[], callback=my_callback):
    await main()

# Callback listener detaches once you exit the context

[garrethcain]

Thanks for the help @puddly

Using your first example, would I add the listener when the device_initialized gets been called?

Eg.

def device_initialized(self, device, *, new=True):
    if device.some_var == "some_value":
        ep = dev.endpoints[1]
        ep.on_off.add_listener(MyListener(ep))

This would allow me to add different/bespoke listeners to specific devices, I'm hoping.

[garrethcain]

Hey @puddly

I think I've done what you said, but it isn't behaving the way I expected so either I don't understand the workflow or something is wrong.

    def device_initialized(self, device, *, new=True):
        """
        Called at runtime after a device's information has been queried.
        I also call it on startup to load existing devices from the DB.
        """
        print("Device is ready: new=%s, device=%s", new, device)

        if device.is_initialized:
            if device.nwk == 0xF0D5:  # Dimmer.
                ep = device.endpoints[1]
                ep.on_off.add_listener(MyDeviceListener(ep))

I added a listener on the on_off cluster for my dimmer. Expecting that when I press on/off it will trigger in the listener. Except the listener is never called.

My listener looks like this;

class MyDeviceListener:
    def __init__(self, endpoint):
        self.endpoint = endpoint
        self.device = endpoint.device

    def cluster_command(self, tsn, command_id, cmd):
        print(f"Device {self.device} received command {cmd} on endpoint {self.endpoint}")

Any thoughts on where I'm going wrong?