XKNX

A KNX library written in Python

Introduction

Changelog

XKNX Object

Lights / Dimmer

Cover

Switches

Time

Sensors

Binary Sensors

HVAC

Configuration

Home Assistant Plugin

The XKNX Object

Overview

The XKNX() object is the core element of any XKNX installation. It should be only initialized once per implementation. The XKNX object is responsible for:

Initialization

xknx = XKNX(config='xknx.yaml',
            loop=loop,
            own_address=Address,
            address_format=GroupAddressType.LONG
            telegram_received_cb=None,
            device_updated_cb=None,
            rate_limit=DEFAULT_RATE_LIMIT)

The constructor of the XKNX object takes several parameters:

Starting

await xknx.start(state_updater=False,
                 daemon_mode=False,
                 connection_config=None)

xknx.start() will search for KNX/IP devices in the network and either build a KNX/IP-Tunnel or open a mulitcast KNX/IP-Routing connection. start() will take the following paramters

Stopping

await xknx.stop()

Will disconnect from tunneling devices and stop the different queues.

Devices

The XKNX may keep all devices in a local storage named devices. All devices may be accessed by their name: xknx.devices['NameOfDevice']. If XKNX receives an update via KNX GROUP WRITE the device is updated automatically.

Example:

outlet = Outlet(xknx,
                name='TestOutlet',
                group_address='1/1/11')
xknx.devices.add(outlet)

await xknx.devices['TestOutlet'].set_on()
await xknx.devices['TestOutlet'].set_off()

Callbacks

The telegram_received_cb will be called for each KNX telegram received by the XKNX daemon. Example:

import asyncio
from xknx import XKNX

async def telegram_received_cb(telegram):
    print("Telegram received: {0}".format(telegram))

async def main():
    xknx = XKNX(telegram_received_cb=telegram_received_cb)
    await xknx.start(daemon_mode=True)
    await xknx.stop()

# pylint: disable=invalid-name
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

For all devices stored in the devices storage (see above) a callback for each update may be defined:

import asyncio
from xknx import XKNX, Outlet


async def device_updated_cb(device):
    print("Callback received from {0}".format(device.name))


async def main():
    xknx = XKNX(device_updated_cb=device_updated_cb)
    outlet = Outlet(xknx,
                    name='TestOutlet',
                    group_address='1/1/11')
    xknx.devices.add(outlet)

    await xknx.start(daemon_mode=True)

    await xknx.stop()

# pylint: disable=invalid-name
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()