Content Index

• What is a KNX IoT API Server?
• How to approach a KNX IoT API Server development?
• How can we interact with a KNX IoT API Server?
• Client implementation examples

What is a KNX IoT API Server?

The KNX IoT API Server is a specific KNX device that can be used to interact with a KNX installation via a standardized RESTful API interface.

The specifications about the RESTful API list of available calls can be found here

How to approach a KNX IoT API Server development?

Introduction

As any other KNX component, the starting point for development resides in the KNX Specifications, for further references see The KNX Standard v3.0.0, chapter 3_10_4 KNX IoT 3rd Party API.

However, to facilitate the comprehension of the KNX IoT ecosystem extension, including the KNX Information Model (aka: KIM) and the new terminology used to describe the entities in use (aka: Semantics), KNX Association made available a set of tools known as Proof of Concept (aka: PoC).

This PoC is a virtual  KNX IoT API server, it runs via docker and can be used for a set of purposes, all intended to reduce the real device time to market, by means of:

• Being a reference, it is possible to make comparisons between it and your own server implementation (eg: for a specific API call result).
• Low down certification efforts, as it comes from KNX it implements the state of the art compliant with the certification requirements for a server device.
• Helping design and validate the Client solutions, meanwhile the hardware design of the Server gets progresses.

More information on how to get the latest version of the PoC and how to run it, can be found here. This is currently a protected resource, to access it please use the following credentials:

Username: knxiotdoc@knx.org

Password: KNX1sgre@t-

How can we interact with a KNX IoT API Server?

The client side of the communication from/to a 3^rd Party KNX IoT API Server relies on two main technics:

• RESTful API Interface
• WebSockets

The full list of supported methods about the RESTful API Interface can be found here. Please consult at any time the schema page here, section KNX IoT 3^rd Party API, chapter Releases for newer versions.

Generally, there is a lot of literature over the internet about RESTful API clients, so every developers’ favorite source code language should find a corresponding library or snippet to perform any of the RESTful API call the server must expose to comply to the KNX IoT standard.

However, there are a list of prerequisites to consider, to ensure that the interaction with a KNX IoT 3^rd Party API Server runs as it should:

• The KNX IoT 3rd Party API Server must be reachable over a certain URL: it sounds basic as concept, but modern network infrastructures are sometimes tricky and it might be the chosen library has some fine tuning to be performed before to start coding on top.
• The KNX IoT 3rd Party API Server must be loaded with a semantic turtle file representing your KNX installation. How to generate this file and how to load it, it’s explained in the PoC documentation, find references in the Server’s chapter at 2.1.
• The KNX IoT 3rd Party API Server must be running: as in the PoC delivery the Server runs via Docker, perform initial checks this environment is correctly up and running.
• In case the KNX IoT 3rd Party API Server has security enabled, then consider to equip your code to support standard OAuth2 authentication mechanisms or run the necessary steps beforehand to get a valid access token for the Client application.

Client implementation examples

Postman collection

The PoC version delivery comes always equipped with a collection of calls ready to be used by Postman. To start getting experience, download and import the given Json file as first step. At this point you must see the details as in Figure 1:

Figure 1

The given collection is structured in folders, organized by topic, to better find the RESTful API call definition desired.

To perform a very first call, some basic settings must be changed, to address the KNX IoT 3^rd Party Server instance in use, find these as shown in Figure 2:

Figure 2

The parameters that need to be changed are the ones highlighted in Figure 3:

Figure 3

Those need to be set to the URL whereas the KNX IoT 3^rd Party Server can be found, on the local LAN or over the Internet. The Ports parameters must be set according to the Server’s settings, here in the example they are bound to default values of the PoC.

Needless to say, once these parameters are set according to Server’s specifications and all the prerequisites mentioned above are satisfied, you just need to run a very first call to interact with the KNX IoT 3^rd Party API Server, as shown in Figure 4:

Figure 4

RESTful python sample call

Here’s a code snippet using Python code to perform a similar call as reported in the example from Postman usage chapter above:

import ssl

import json

import time

import urllib3

import requests

import sseclient

import threading

 

class RESTfulClient:

  #Constructor

  def __init__(self, options = {}):

    # Private attributes

    self.instance = 0

    self.configs = None

    self.clientid = None

    self.clientsecret = None

    self.tokenurl = None

    self.tokenjson = None

    self.lateststatus = None

    self.internalq = None

    # Parse constructor variables into attributes

    for key in options:

      if key.lower() == "instance":

        self.instance = int(options[key])

      if key.lower() == "clientid":

        self.clientid = options[key]

      if key.lower() == "clientsecret":

        self.clientsecret = options[key]

      if key.lower() == "tokenurl":

        self.tokenurl = options[key]

      if key.lower() == "tokenjson":

        self.tokenjson = options[key]

  # Public method refresh

  def refresh(self, remotableflag, genericonoffflag, failureflag, remainingtime):

    if (self.lateststatus is not None):

      return 0

    try:

      token_url = self.tokenurl

      state_api_url = "http://127.0.0.1:5000/.well-known/knx"

      #Client (application) credentials retrieved by KNX 3rd Party API PoC authentication flow

      client_id = self.clientid

      client_secret = self.clientsecret

 

      tokens = json.loads(self.tokenjson)

      access_token_value = tokens['access_token']

 

      api_call_headers = {'Authorization': 'Bearer ' + access_token_value}

      api_call_response = requests.get(state_api_url, headers=api_call_headers, verify=False)

      #print(api_call_response.text)

      result = json.loads(api_call_response.text)

    except:

      print("RESTful API Error")

      return 1

    

    self.lateststatus = "Status: "

    for key, value in result.items():

      if (key == "version"):

        print("version = %s" % value)

        self.lateststatus = self.lateststatus + "Version = " + str(value)

      if (key == "..."):

        print("... = %s" % value)

 

    return 0

  # Public method getter attribute >lateststatus

  def get_lateststatus(self):

    return self.lateststatus

  # Public method stop

  def stop(self):

    return 0
 

  #Destructor

  def __del__(self):

    return 0

The class is taking as constructor’s parameters the following values coming from a config file:

restfulapi:

  client_id: 'd681a3bd-b87a-4ab2-8822-b7979c7fadbd'

  client_secret: '3282dd68-5fb8-49e6-b1c8-e109ad9b18e5' 

  auth_url: 'http://127.0.0.1:8083/auth/realms/knx/protocol/openid-connect/auth?response_type=code&client_id=knxapi&redirect_uri=http://127.0.0.1:5000/test-oidc-callback&scope=read%20write%20manage%20time_3600'

  token_url: 'http://127.0.0.1:8083/oauth/access'

  auth_flow: 'authorizationcodeflow'

  account_username: 'user_foo'

  account_password: 'user_foo'

  tokenjson: '{"access_token": "DE_b383d9f06065c9c866cd62dbd4f4b863", "refresh_token": "DE_88bdb3faaf043d8ccc9ba1af7d30325d", "token_type": "Bearer", "expires_in": 2592000}'

as they are specified in the KNX specifications and presented in the KNX IoT 3^rd Party API calls collection page.

Some remarks:

• The authentication flow runs outside the presented code, to execute the call loaded in the variable state_api_url, the tokenjson parameter in the config file must already be filled in with a valid token exchanged with the Server. Use Postman collection, folder Login Flow, in case you need to keep the OAuth tasks separated from your Client application scope.

WebSockets sample class

Here’s a code snippet using Python code to perform a WebSocket interaction to subscribe to KNX status changes on predefined project’s element:

import json

import time

import select

import signal

import urllib3

import asyncio

import requests

import threading

import websockets

 

class KNX:

  #Constructor

  def __init__(self, options = {}):

    # Private attributes

    self.__asyncioparentloop = None

    self.__syncsemaphore = None

    self.instance = 0

    self.configs = None

    self.wsurl = None

    self.wshconnection = None

    self.wshugrade = None

    self.wshsecwebsocketkey = None

    self.wshsecwebsocketprotocol = None

    self.wshsecwebsocketversion = None

    self.wsid = None

    self.lateststatus = None

    self.monitorloop = False

    self.monitor = None

    # Parse constructor variables into attributes

    for key in options:

      if key.lower() == "instance":

        self.instance = int(options[key])

      if key.lower() == "wsurl":

        self.wsurl = options[key]

      if key.lower() == "wshconnection":

        self.wshconnection = options[key]

      if key.lower() == "wshugrade":

        self.wshugrade = options[key]

      if key.lower() == "wshsecwebsocketkey":

        self.wshsecwebsocketkey = options[key]

      if key.lower() == "wshsecwebsocketprotocol":

        self.wshsecwebsocketprotocol = options[key]

      if key.lower() == "wshsecwebsocketversion":

        self.wshsecwebsocketversion = options[key]

 

  # Private coroutine for WebSocket connection handling

  async def __wsconnect(self, uri, sp):

    try:

      self.wsid = await websockets.connect(uri, subprotocols=sp)

    except:

      self.wsid = None

 

  # Private coroutine for WebSocket connection handling

  async def __wsdisconnect(self):

    try:

      await self.wsid.close()

    except:

      self.wsid = None

 

  # Private coroutine for WebSocket messaging handling

  async def __wsread(self):

    try:

      self.lateststatus = await self.wsid.recv()

    except:

      self.lateststatus = None

 

  # Private coroutine for WebSocket messaging handling

  async def __wssend(self, msg):

    rcode = 0

    try:

      await self.wsid.send(msg)

    except:

      rcode = 1

    return rcode

  

  # Private thread routine for WebSocket monitor

  def __wsmonitor(self, ploop):

    while (self.monitorloop == True):

      #As this is a different thread, we need to bind to the same parent event loop and coordinate via a semaphore (not real-time performance, but can be better plaing down with timeout parameter...)

      self.__syncsemaphore.acquire()

      ploop.run_until_complete(asyncio.wait_for(self.__wsread(), 1))

      self.__syncsemaphore.release()

      if (self.lateststatus is not None):

        print("From KNX WebSocket: <%s>" % str(self.lateststatus))

        #Add relevant content to the internal queue to be propagated towards other appliances or system connected via Cloud

      else:

        print("From KNX Websocket: Timeout, retry listen...")

    return 0

 

  # Public method authenticate

  def authenticate(self):

    #Connect to the PoC WebSockets channel

    loop = asyncio.get_event_loop()

    loop.run_until_complete(asyncio.wait_for(self.__wsconnect(self.wsurl, [self.wshsecwebsocketprotocol]), 5))

    #Propagate the sync necessary entities

    self.__syncsemaphore = threading.Semaphore(1)

    self.__asyncioparentloop = loop

    if (self.wsid is not None):

      #Start a monitor thread to read all the traffic over the PoC WebSocket channel

      self.monitorloop = True

      self.monitor = threading.Thread(target=self.__wsmonitor, args=[self.__asyncioparentloop])

      self.monitor.start()

      #Subscribe for KNX event changes on dedicated integration object (well known upfront in this code)

      genericonoff='{"meta": {"transaction": {"id": "cs2","type": "org.knx.gateway.subscribe","source": "/topic/client/s2"}},"data": {"type": "subscription","relationships": {"subscriptionDatapoints": {"data": [{"id": "b62f4a5c-dcb7-4084-9332-668da855ca81","type": "datapoint"}]}}}}'

      self.__syncsemaphore.acquire()

      self.__asyncioparentloop.run_until_complete(asyncio.wait_for(self.__wssend(genericonoff), 5))

      self.__syncsemaphore.release()

      remainingtime='{"meta": {"transaction": {"id": "cs4","type": "org.knx.gateway.subscribe","source": "/topic/client/s4"}},"data": {"type": "subscription","relationships": {"subscriptionDatapoints": {"data": [{"id": "c93749cb-49eb-4104-a764-12fb915170ba","type": "datapoint"}]}}}}'

      self.__syncsemaphore.acquire() 

      self.__asyncioparentloop.run_until_complete(asyncio.wait_for(self.__wssend(remainingtime), 5))

      self.__syncsemaphore.release()

 

    return 0

 

  # Public method getter attribute >wsid

  def get_wsid(self):

    return self.wsid

 

   # Public method getter attribute >lateststatus

  def get_lateststatus(self):

    return self.lateststatus

 

  # Public method stop

  def stop(self):

    # Stop monitor thread

    self.monitorloop = False

    if (self.monitor is not None):

      self.monitor.join()

    # Cleanly close the websocket connection towards the PoC

    if (self.wsid is not None):

      self.__asyncioparentloop.run_until_complete(asyncio.wait_for(self.__wsdisconnect(), 5))

    return 0

 

  #Destuctor

  def __del__(self):

    return 0

The class is taking as constructor’s parameters the following values coming from a config file:

knx:

  wsurl: 'ws://127.0.0.1:5000/api/v2/messaging/ws'

  wshconnection: 'Upgrade'

  wshupgrade: 'websocket'

  wshsecwebsocketkey: 'S05YQUZvckV2ZXIuLi4uLg=='

  wshsecwebsocketprotocol: 'gw.knx.org'

  wshsecwebsocketversion: '13'

  wssubscriptiondelaycommand: ''

As they are specified in the KNX specifications and presented in the KNX IoT 3^rd Party API calls collection page.

Some remarks:

• As variable wsurl is set to 127.0.0.1 the assumption is to run this Client code class on the same machine where a Server instance is up and running.
• The entry offset to read the code shall be set to the authenticate method, assuming an external piece of code instantiated the class via its constructor using the parameters reported above.
• There is code overhead to make more explicit to the readers some details about the inter process communication from the typical Python way to use the Async domain, whereas those details are hidden. This effects the style only, not the concepts.
• The goal of the code is to give a syntax example of what is needed to operate via a subscription, assuming the datapoints ID as known and valid in the loaded project scope in the running Server: these IDs are the first to be changed to make this snippet to run at the reader’s premises.
• The expected result behavior of the proposed code is to receive from the Server notification of events happening on the datapoints located at the given IDs, such as like “The kitchen light is now ON” or “The temperature in the Studio is now 28,5 °C”. What to do after the Client receives these notification of events, it’s pure matter of the Client application.
• There are other interactions possibilities via the WebSocket channel, such as trigger commands to a specific datapoint. The code structure fits, what needs to be changed is the payload exchange over the WebSocket channel towards the server. As payload formatted example given:

{

  "meta": {

    "transaction": {

      "id": "c2",

      "type": "org.knx.gateway.write",

      "source": "/topic/client/2"

    }

  },

  "data": [

    {

      "id": "b62f4a5c-dcb7-4084-9332-668da855ca81",

      "type": "datapoint",

      "attributes": {

        "value": "on"

      }

    }

  ]

}

where the expected result is to “Turn Kitchen light ON”