Sending KNX telegrams via appdaemon

Hello guys,

whats the right syntax for calling a service from appdaemon?

This is what the service cal would look like in an automation:

  action:
    - service: knx.send
      data:
        address: '1/3/9'
        payload:
          - 255

How would that look like in app daemon? I tried this and got an error message

self.call_service("knx/send", data = {'address': '1/3/9', 'payload': [255]})

Can you please post the error message? Can you confirm that the automation you posted works?

Confirmative, the automation works.

The error is:

2020-06-16 15:42:32.604838 WARNING HASS: Code: 400, error: 400: Bad Request
2020-06-16 16:02:56.027222 WARNING HASS: Error calling Home Assistant service default/knx/send

I don’t own any KNX devices, but I once helped someone with KNX lights. I assume the service you try to call here is turning on the light at full brightness? May I ask why you are not using the KNX light integration?

Do other service calls from AppDaemon work?
Did you try with a simple payload of 255 instead of the list [255]?

If you pass an int to the payload it will be parsed as a binary (actually max. 6bit value Encoded in a Knx Frame used only for DPT 1, 2 and 3).
A percent value should be passed as an iterable (255,).

1 Like

Hello,

thanks for chiming in. I need additional help. I am sorry. How would that look like in Python code? Think of this …

        data = {"address": self.args["position_address"], "data": [int( 255.0 * self.currentPosition )]}
        self.log("setting Position %f" %(self.currentPosition))
        self.call_service("knx/send", data = data)

And how does it look like when sending a short/long commands?

        data = {"address": self.args["move_short_address"], "data": [1]}
        self.call_service("knx/send", data = data)
        data = {"address": self.args["move_long_address"], "data": [0]}
        self.call_service("knx/send", data = data)

Why do I need this? I do have a cover KNX actor which does not have a position address. I wrote some python appdaemon code which takes care of this by listening on the KNX interface to calculate the position of the cover and updates the HA cover element. And yes, this is already working when your KNX actor has a position element, but my actor hasn’t. That’s the only way to make it work.

As soon this works I will publish the appdaemon class. All the calculation works in a few lines of code. I am surprised by myself how easy it seemed. Only puzzle piece missing is sending the information and commands to the KNX bus.

Regards.
Ralf

Apart from me not knowing what appdaemon is, I can say that you probably shouldn’t send the move_short_address and move_long_address data as a list. These are DPT1 communication objects so (see my previous post) their data should be integer to encode it correctly.

Where do you send your newly calculated position to on the knx bus? Don’t you want this information in HA?

This is already implemented in xknx - its called travel calculator. Maybe you can use it to add the feature to xknx directly.

@farmio. See Issue #25. Isn’t it you answering in that thread to me?

I do not feel literate enough in addons/python to do that. I am happy to learn fast … . Wrote today my 1st appdaemon app which handles all of this. Only sending the information back to KNX (which is read back by HA and available again in HA).

I will look into the travel calculator. As I said. It works for covers which have the position address, If this does not exist it’s not working. No bus activity is updating the HA cover element without my code.

There seems to a missing link between the BUS and the KNX cover item. See this thread.

Yes thats me too.

If you send to a GA from HA I’m not sure if xknx doesn’t ignore it because its sent by itself.
If you didn’t set a position_address then there isn’t even a GA for this to receive so this should probably lead to an error.

I think you would be better off using a template cover in HA and updating its state from knx - or do the calculation in the knx integration itself.

Sad, I am so close to finishing this. What I do not understand in the integration where to look for is the lines of code which connect the KNX bus with xknx and travel calculator.

My appdaemon code is pretty straight forward. I have put in the cover element and KNX address. However not the KNX actor is handling this, but the app daemon code.

Let me share what I have so far for others to make use of it. Perhaps we can make this happen. There must be somewhere an example of a successful usage of the ´knx.send´ service from within appdaemon.

Here is the change in the knx section

knx:
  ...
  expose:
    - type: 'percentU8'
      entity_id: cover.arbeiten
      attribute: position
      address: '1/3/8'
  ...
  fire_event: true
  fire_event_filter:
    - "1/3/1-6"     # Rolladen Ergeschoss Lang/Kurz Objekte
    - "2/3/1-4"     # Rolladen Obergeschoss Lang/Kurz Objekte
    - "1/3/7-12"    # Rolladen Ergeschoss Status Objekte
    - "2/3/9-12"    # Rolladen Obergeschoss Status Objekte

Here is the knx cover item

  - platform: knx
    name: "Arbeiten"
    move_short_address: '1/3/1'
    move_long_address: '1/3/2'
    position_address: '1/3/7'       # My actor has now position adress.
    position_state_address: '1/3/8' # Same here. Both of these will be taken care by app daemon
    travelling_time_down: 12
    travelling_time_up: 12

Here is my appdaemon app in file cover_update_position.py

import appdaemon.plugins.hass.hassapi as hass
import json
from datetime import datetime, timedelta

class CoverUpdatePosition(hass.Hass):

    def initialize(self):
        self.listen_event(self.on_state_change, "knx_event", address = self.args["move_long_address"])
        self.listen_event(self.on_stop,         "knx_event", address = self.args["move_short_address"])
        self.listen_event(self.on_set_position, "knx_event", address = self.args["position_address"])

        self.currentPosition     = 1.0                # 0.0 to 1.0 representing 1.0 is top, 0.0 is botton
        self.timestampLastChange = self.datetime()    # ts in milliseconds of last Motion
        self.directionLastMotion = 0                  # Direction -1 = down; +1 = up

    # The following 5 routines take care of all updating the self.currentPosition by
    # just listening to the STOP/UP/DOWN command on the knx bus

    def on_state_change(self, event_name, data, *kwargs):
        self.update_position()
        self.directionLastMotion = -1 if data['data'] == 1 else +1 
        self.handle = self.run_in(self.timeout_handler, self.args["travelling_time"] + 2)

    def on_stop(self, event_name, data, *kwargs):
        self.update_position()
        self.directionLastMotion = 0 

    def timeout_handler(self, *kwargs):
        self.update_position()
        self.directionLastMotion = 0

    def update_position(self):
        self.currentPosition = (
            max(
                min(
                    1.0, 
                    self.currentPosition + 
                        self.directionLastMotion * 
                        (self.datetime() - self.timestampLastChange).total_seconds() / 
                        self.args["travelling_time"]
                ), 
                0.0)
            )
        self.timestampLastChange = self.datetime()
        self.log('Updating position to %f ' % (self.currentPosition))
        self.cover_set_position()

    # Here the newly calculated position is sent to the KNX position adress.
    # the actuall knx.send service is not working yet

    def cover_set_position(self, *kwargs):
        data = {"address": self.args["position_address"], "data": [int( 255.0 * self.currentPosition )]}
        self.log("setting Position %f" %(self.currentPosition))
        # self.call_service("knx/send", data = data)

    # The following 4 routines take care for setting the position to be translated
    # into proper UP/DOWN/STOP commands to respect the requried changes.

    def on_set_position(self, event_name, data, *kwargs):
        desiredPosition  = data['data'][0] / 255
        desiredDirection = +1 if desiredPosition - self.currentPosition > 0 else -1
        desiredTime      = abs(desiredPosition - self.currentPosition) * self.args["travelling_time"]
        self.log('sending SET notification with (from %f to %f direction %d time %f)' % (self.currentPosition, desiredPosition, desiredDirection, desiredTime))
        self.cover_down() if desiredDirection == -1 else self.cover_up()
        self.handle = self.run_in(self.cover_stop, desiredTime)

    # Here the UP/DOWN/STOP commands is sent to the KNX position adress.
    # the actuall knx.send service is not working yet

    def cover_stop(self, *kwargs):
        self.log('send STOP')
        data = {"address": self.args["move_short_address"], "data": [1]}
        # self.call_service("knx/send", data = data)

    def cover_up(self, *kwargs):
        self.log('send UP')
        data = {"address": self.args["move_long_address"], "data": [0]}
        # self.call_service("knx/send", data = data)

    def cover_down(self, *kwargs):
        self.log('send DOWN')
        data = {"address": self.args["move_long_address"], "data": [1]}
        # self.call_service("knx/send", data = data)

and this is the actual definition of the app in apps.yaml.

cup_arbeiten:
  module: cover_update_position
  class: CoverUpdatePosition
  move_short_address: '1/3/1'
  move_long_address: '1/3/2'
  position_address: '1/3/7'
  travelling_time: 12

I hope someone finds this useful and helps me to get this working. I am happy to assist.

just a wild guess: try using service registry to call the service


self.services.call("knx", "send", service_data=data) when your self here is a HomeAssistant class.

Hi @rak, were you able to solve that problem? I am currently experiencing the same error.
Any help would be appreciated.

Thanks!

No. I stopped using this solution.

For those who are looking for a solution: Unable to call_service knx/send · Issue #1754 · AppDaemon/appdaemon · GitHub

self.call_service("knx/send", address="16/0/0", payload="Hello KNX", type="16.000")