[Help] - zigbee readout about zigpy-znp

[puddly]

zigpy is stable. zigpy-znp is also somewhat stable but since few people use it right now, I can't guarantee you won't encounter any bugs.

[Adminiuga]

There's no documentation for things like that. You would need to write your own application on top of zigpy-znp similar to bellows cli implementation

[puddly]

Using zigpy directly is unfortunately undocumented but the interface will be the same regardless of what radio hardware you use. This code is completely untested and probably won't work but it should get you started:

import asyncio
import logging
import coloredlogs

import zigpy_znp.types as t
import zigpy_znp.commands as c

# There are many different radio libraries but they all have the same API
from zigpy_znp.zigbee.application import ControllerApplication

coloredlogs.install(level=logging.DEBUG)
logging.getLogger('zigpy_znp').setLevel(logging.DEBUG)


LOGGER = logging.getLogger(__name__)


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):
        asyncio.create_task(self.async_device_joined(device))

    async def async_device_joined(self, device):
        LOGGER.info("Device joined: %s", device)

        for endpoint_id, endpoint in device.endpoints.items():
            # If it's a proper motion sensor, it probably has an IAS Zone cluster
            # that you need to bind to before it does anything.
            #
            # If it's a Xiaomi motion sensor, it'll just send attribute updates.
            if not hasattr(endpoint, 'ias_zone'):
                continue

            endpoint.ias_zone.add_context_listener(self)

            await endpoint.ias_zone.bind()
            await endpoint.ias_zone.write_attributes({'cie_addr': self.application.ieee})
            await endpoint.ias_zone.enroll_response(0x00, 0x00)

    def attribute_updated(self, device, cluster, attribute_id, value):
        # This is where you'll receive attribute updates
        LOGGER.info("Received an attribute update %s=%s on cluster %s from device %s", 
            attribute_id, value, cluster, device)



async def main():
    app = ControllerApplication(ControllerApplication.SCHEMA({
        'database_path': '/path/to/zigbee.db',
        'device': {
            'path': '/dev/serial/by-id/your-serial-port',
        }
    }))

    listener = MainListener(app)
    app.add_listener(listener)

    await app.startup(auto_form=True)

    # Permit joins for a minute
    await app.permit(60)
    await asyncio.sleep(60)

    # Just run forever
    while True:
        asyncio.sleep(10000)


if __name__ == '__main__':
    await asyncio.run(main())

[Adminiuga]

some primer on zigbee https://github.com/MarkDing/IoT-Developer-Boot-Camp/wiki/Zigbee-Boot-Camp

[Mopele]

Ok thanks for the great info! So basically this module is still in an early phase of development and not quite suited for my needs right?

[puddly]

I'm closing this issue for now since it doesn't directly relate to zigpy-znp, but feel free to comment if you need any help using zigpy on its own.

[Hedda]

@Mopele Please consider contributing to the discussions in zigpy/zigpy#471 and zigpy/zigpy#477