Code samples

[naive-HA]

Are there any code samples showing how to use zigpy-znp standalone?
I am writing my own code for raspberry pi zero 2W: not making use of Home Assistant or alike.
I am writing my own python code and would like to grow my code base with code interfacing with zigbee devices.
I have a sonoff usb 3.0 and zigpy-znp seems the solution upon I could build my code
Any tips on how/where to start?

Any guidance is much appreciated

[naive-HA]

Anyone?

[puddly]

Zigpy at the moment is lacking documentation but there are code snippets referenced in this issue that may help: zigpy/zigpy#715 (comment)

[naive-HA]

@puddly thank you

[naive-HA]

I would appreciate some further help here:
I have a Sonoff usb 3.0 and a button which I have paired and made sure all works fine: I have checked with Home Assistant
I have made use of the code sample in: zigpy/zigpy#452 (comment)
My Sonoff device is initialized and the button connects. I get the message:
Device joined: <Device model=None manuf=None nwk=0xD6C5 ieee=00:12:4b:00:24:54:94:f9 is_initialized=False>
The message is printed by the device_joined method of MainListener class
Then things turn ugly: When I press the button first time, the following message is displayed
Error calling listener <bound method ClusterPersistingListener.attribute_updated of <zigpy.zcl.ClusterPersistingListener object at 0x772dcb28fe80>> with args (4, 'eWeLink'): AttributeError("'NoneType' object has no attribute 'attribute_updated'")
which I understand is triggered as the method listener_event of ListenableMixin class in util.py is looking for the attribute_update method of the wrong object.
First question: Am I doing something wrong? Am I making use of an obsolete code?

Then, pressing the button further delivers nothing. No warning, no error, no message at all.
So, I have implemented a new method handle_message for class MainListener

class MainListener:
    """
    Contains callbacks that zigpy will call whenever something happens.
    Look for `listener_event` in the Zigpy source or just look at the logged warnings.
    """

    def __init__(self, application):
        self.application = application

    def device_joined(self, device):
        print(f"Device joined: {device}")

    def attribute_updated(self, device, cluster, attribute_id, value):
        print(f"Received an attribute update {attribute_id}={value}"
              f" on cluster {cluster} from device {device}")

    def handle_message(self, sender, profile, cluster, src_ep, dst_ep, message):
        print(f"sender: {sender}; profile: {profile}; cluster: {cluster}; src_ep: {src_ep}; dst_ep: {dst_ep}; message: {message}")

Now, when I restart the script, I get all the above messages, but if I continue to press the button, the new method is being called and I can see messages coming through, like:
b'\x01\xD3\x01'
where the last position: x01 is double click, the middle position is a counter and the first position I believe is describing a tap/click
Second question is: is there a better way of doing this?

thx in advance

[naive-HA]

just for completeness: a single click is x02, double click is x01 and long click is x00

[naive-HA]

problem solved

[Hedda]

@naive-HA You might for reference also want to look at the code of the Zigbee Plugin for Domoticz (Beta branch), and the Zigbee Plugin for Jeedom (competing open-source home automation software) are all using zigpy libraries as dependencies with and without the main zigpy library:

https://github.com/zigbeefordomoticz/Domoticz-Zigbee/

PS: Note that to get to code of Zigbee Plugin for Jeedom you would actually need to buy and install that plugin in Jeedom (which once installed will give you access to Python code of all installed plugins) as they, unfortunately, do not actually distribute their Zigbee plugin code directly via a Git repository.

[naive-HA]

@Hedda your help is appreciated

[Hedda]

@naive-HA By the way, the latest code for Zigbee Plugin for Domoticz is only in their Beta (beta6) branch:

https://github.com/zigbeefordomoticz/Domoticz-Zigbee/tree/beta6

[naive-HA]

Hopefully no one will struggle like me:

#coordinator: SONOFF Zigbee 3.0 USB dongle
#device: SONOFF temperature sensor Device model='TH01' manuf='eWeLink'
import sys, asyncio
import zhaquirks #how to use quirks?
import zigpy.profiles
from zigpy_znp.zigbee.application import ControllerApplication


class MainListener:
    def __init__(self, application):
        self.application = application
        self.ieee = list()

    def device_joined(self, device):
        print(f"Device joined: {device}")
        self.ieee.append(device.ieee)

    def device_left(self, device):
        print(f"Device left: {device}")
        if device.ieee in self.ieee:
            self.ieee.remove(device.ieee)

    def device_removed(self, device):
        print(f"Device removed: {device}")
        if device.ieee in self.ieee:
            self.ieee.remove(device.ieee)

    def device_initialized(self, device, *, new=True):
        if device.nwk == 0x0000:
            return #skip coordinator
        for endpoint_id, endpoint in device.endpoints.items():
            if endpoint_id == 0:
                continue  #skip zdo

            # You need to attach a listener to every cluster to receive events
            for cluster in endpoint.in_clusters.values():
                # The context listener passes its own object as the first argument
                # to the callback
                cluster.add_context_listener(self)

            for cluster in endpoint.out_clusters.values():
                # The context listener passes its own object as the first argument
                # to the callback
                cluster.add_context_listener(self)

            print("listing all attributes and commands for each cluster of each end_point")
            print(getattr(endpoint, 'device_type', None))
            print("output_clusters:", [cluster.cluster_id for cluster in endpoint.out_clusters.values()])
            for cluster_id in [cluster.cluster_id for cluster in endpoint.out_clusters.values()]:
                for attribute in list(device.endpoints[endpoint_id].out_clusters[cluster_id].attributes_by_name):
                    print("cluster: ", cluster_id, ": ", device.endpoints[endpoint_id].out_clusters[cluster_id].ep_attribute, \
                        " :: attribute: ", attribute)
                for command_id in list(device.endpoints[endpoint_id].out_clusters[cluster_id].server_commands):
                    print("cluster: ", cluster_id, ": ", device.endpoints[endpoint_id].out_clusters[cluster_id].ep_attribute, \
                        " :: command: ", device.endpoints[endpoint_id].out_clusters[cluster_id].server_commands[command_id].name)
            print(30*"X")
            print("input_clusters:", [cluster.cluster_id for cluster in endpoint.in_clusters.values()])
            for cluster_id in [cluster.cluster_id for cluster in endpoint.in_clusters.values()]:
                for attribute in list(device.endpoints[endpoint_id].in_clusters[cluster_id].attributes_by_name):
                    print("cluster: ", cluster_id, ": ", device.endpoints[endpoint_id].in_clusters[cluster_id].ep_attribute, \
                        " :: attribute: ", attribute)
                for command_id in list(device.endpoints[endpoint_id].in_clusters[cluster_id].server_commands):
                    print("cluster: ", cluster_id, ": ", device.endpoints[endpoint_id].in_clusters[cluster_id].ep_attribute, \
                        " :: command: ", device.endpoints[endpoint_id].in_clusters[cluster_id].server_commands[command_id].name)
            print(30*"X")

    def attribute_updated(self, cluster, attribute_id, value):
        print(f"Received an attribute update "
              f"{cluster._endpoint._device.endpoints[cluster._endpoint._endpoint_id].in_clusters[cluster.cluster_id].find_attribute(attribute_id).name}={value}"
              f" on cluster {cluster.ep_attribute} from device {cluster._endpoint._device._ieee}")

    def cluster_command(self, hdrtsn, hdrcommand_id, args): #not sure what this does
        print(f"command ID: {hdrcommand_id}")

async def main():
    #for Debian use --> ls -l /dev/serial/by-id
    #for Windows use --> "path": "COM3" or whatever Device Manager shows for the coordinator
    app = await ControllerApplication.new(config=ControllerApplication.SCHEMA({
            "database_path" : "/tmp/zigbee" + str(int(time.time())) + ".db",
            "device"        : {"path": "/dev/serial/by-id/usb-ITead_Sonoff_Zigbee_3.0_USB_Dongle_Plus_b47ef8a90460ec119f99345f25bfaa52-if00-port0"}}),
        auto_form=True,
        start_radio=True,
    )   
    
    listener = MainListener(app)
    app.add_listener(listener)
    # Have every device in the database fire the same event so you can attach listeners
    for device in app.devices.values():
        listener.device_initialized(device, new=False)

    permit_duration = 20
    await app.permit(permit_duration)
    print("join now")
    await asyncio.sleep(permit_duration)
    print("joining ends")

    for device in app.devices.values():
        # if str(device.ieee) == "00:12:4b:00:25:15:98:f3":
        if device.nwk == 0x0000:
            continue #skip coordinator
        await device.endpoints[1].in_clusters[1026].read_attributes(["measured_value"])
        await device.endpoints[1].in_clusters[1029].read_attributes(["measured_value"])

        #ep_attribute = "identify"
        #server_commands: dict[int, ZCLCommandDef] = {0x00: ZCLCommandDef("identify", {"identify_time": t.uint16_t}, False),
        # await device.endpoints[1].identify.identify(10)
        await device.endpoints[1].in_clusters[3].identify(10) #same same

        await device.zdo.bind(device.endpoints[1].in_clusters[1026])
        #async def configure_reporting(self, attribute: int | str, min_interval: int, max_interval: int, reportable_change: int, manufacturer: int | None = None,
        #device reports temperature 22.22 C degrees as 2222, thus reportable change 1 is 0.01 C deg
        await device.endpoints[1].in_clusters[1026].configure_reporting('measured_value', 0, 3, 1)
        
        await device.zdo.bind(device.endpoints[1].in_clusters[1029])
        #device reports humidity 88.88% as 8888, thus reportable change 1 is 0.01%
        await device.endpoints[1].in_clusters[1029].configure_reporting('measured_value', 0, 3, 1)
        await asyncio.sleep(20)

        print("reporting stopped")
        #stop reporting
        await device.endpoints[1].in_clusters[1026].configure_reporting('measured_value', 4, 3, 1)
        await device.endpoints[1].in_clusters[1029].configure_reporting('measured_value', 4, 3, 1)

        await asyncio.sleep(20)
        print("restting to factory settings")
        #0x00: ZCLCommandDef("reset_fact_default", {}, False)
        res = await device.endpoints[1].in_clusters[0].reset_fact_default()
        print(res)

        print("I am yet to discover how to gracefully exit the loop...")

if __name__ == "__main__":
    import time
    asyncio.run(main())

[Hedda]

@naive-HA Since this specific issue is closed and could thus easily be ignored may I suggest that you instead open either new issues -> https://github.com/zigpy/zigpy-znp/issues/new/choose (noting that each issue is usually meant to separate for specific issues for tracking so that is can be closed once answered or fixed/resolved) or open new discussion(s) for little broader ongoing generic dialogue(s) that will not be closed -> https://github.com/zigpy/zigpy-znp/discussions

[Hedda]

@naive-HA Since this specific issue is closed and could thus easily be ignored may I suggest that you instead open either new issues -> https://github.com/zigpy/zigpy-znp/issues/new/choose (noting that each issue is usually meant to separate for specific issues for tracking so that is can be closed once answered or fixed/resolved) or open new discussion(s) for little broader ongoing generic dialogue(s) that will not be closed -> https://github.com/zigpy/zigpy-znp/discussions

@naive-HA For example see that the Zigbee plugin for the Domoticz project started as a discussion here before it grew so large that needed to move to posting specific things as issues for tracking -> zigpy/zigpy#865