started knx integration

CMakeLists.txt

@@ -101,6 +101,16 @@ if(HAVE_MQTT)
   endif(mqtt STREQUAL ON)
 endif(HAVE_MQTT)
 
+find_library(HAVE_KNX eibclient)
+if(HAVE_KNX)
+  option(knx "disable support for KNX handling." ON)
+  if(knx STREQUAL ON)
+    message(STATUS "KNX enabled")
+  else(knx STREQUAL ON)
+    unset(HAVE_KNX)
+  endif(knx STREQUAL ON)
+endif(HAVE_KNX)
+
 find_library(HAVE_SSL ssl)
 find_library(LIB_CRYPTO crypto)
 if(HAVE_SSL)

config.h.cmake

@@ -7,6 +7,9 @@
 /* Defined if MQTT handling is enabled. */
 #cmakedefine HAVE_MQTT
 
+/* Defined if KNX is enabled. */
+#cmakedefine HAVE_KNX
+
 /* Defined if SSL is enabled. */
 #cmakedefine HAVE_SSL
 

configure.ac

@@ -63,6 +63,16 @@ if test "x$with_mqtt" != "xno"; then
 fi
 AM_CONDITIONAL([MQTT], [test "x$with_mqtt" != "xno"])
 
+AC_ARG_WITH(knx, AS_HELP_STRING([--without-knx], [disable support for KNX handling]), [], [with_knx=yes])
+if test "x$with_knx" != "xno"; then
+	AC_CHECK_LIB([eibclient], [EIBSocketURL],
+		[AC_DEFINE_UNQUOTED(HAVE_KNX, [1], [Defined if KNX handling is enabled.])
+		EXTRA_LIBS+=" -leibclient"],
+		[AC_MSG_RESULT([Could not find EIBSocketURL in libeibclient.])
+		with_knx="no"])
+fi
+AM_CONDITIONAL([KNX], [test "x$with_knx" != "xno"])
+
 AC_ARG_WITH(ssl, AS_HELP_STRING([--without-ssl], [disable support for SSL]), [], [with_ssl=yes])
 if test "x$with_ssl" != "xno"; then
 	AC_CHECK_LIB([ssl], [OPENSSL_init_ssl],

contrib/etc/ebusd/knx.cfg

@@ -0,0 +1,70 @@
+# Configuration file for ebusd KNX integration with knxd (https://github.com/knxd/knxd).
+
+# Use this file with ebusd to establish a bridge between KNX and eBUS for a set of messages.
+# The commandline options to ebusd should contain e.g.:
+# --knxurl=ip:localhost --knxint=/etc/ebusd/knx.cfg
+
+# Currently only reading from and writing to group addresses as defined here is supported.
+# Setting the addresses via ETS is not (yet) possible as well as setting the physical address.
+# All entries are set to group address flags as follows:
+# - for read and passive write messages: "Read", "Transmit"
+# - for active write messages: "Write", (no "Update")
+
+# The physical address depends on how knxd is configured and might change each time ebusd and/or knxd is restarted when
+# a range of possible client addresses were configured on knxd side (which is recommended when there is more than one
+# client using knxd).
+
+
+# the global value group assignments for running, version, signal, uptime, updatecheck, and scan.
+# running: 1 bit, 1=running, DPT 1.002
+global/running = 9/0
+# version: 2 octets, major in MSB, minor in LSB, DPT 217.001 "DPT_Version"
+global/version = 9/1
+# signal: 1 bit, 1=signal acquired, DPT 1.002
+global/signal = 9/2
+# uptime: 4 octets int, seconds since start, sent once every hour, DPT 12.100
+global/uptime = 9/3
+# updatecheck: 1 bit, 1=update available, DPT 1.002
+global/updatecheck = 9/4
+# scan: 1 bit, 1=running, DPT 1.002
+global/scan = 9/5
+
+
+# the message field value group assignments by circuit/message/field name.
+# the value coding depends on the field datatype and currently only numeric datatypes are supported.
+# the mapping is as follows:
+# - BI0:1 - BI7:1, length 1: 1 bit, DPT 1
+# - without divisor:
+#   - BI0 - BI6, length >1: 1 octet, unsigned, DPT 5.010
+#   - UCH: 1 octet, unsigned, DPT 5.010
+#   - SCH, D1B: 1 octet, signed, DPT 6.010
+#   - UIN, UIR, PIN: 2 octet, unsigned, DPT 7.001
+#   - SIN, SIR: 2 octet, signed, DPT 8.001
+#   - U3N, U3R, ULG, ULR: 4 octet, unsigned, DPT 12.001
+#   - U3N, U3R, SLG, SLR: 4 octet, signed, DPT 13.001
+# - with divisor:
+#   - BI0 - BI6, length >1: 2 octet, signed float, DPT 9.*
+#   - UCH, SCH, D1B, UIN, UIR, SIN, SIR: 2 octet, signed float, DPT 9.*
+#   - U3N, U3R, ULG, ULR, SLG, SLR: 4 octet, signed float, DPT 14.*
+# - with or without divisor:
+#   - D1C, D2B, D2C, FLT, FLR: 2 octet, signed float, DPT 9.*
+#   - EXP, EXR: 4 octet, signed float, DPT 14.*
+#
+# note: the float conversion from ebus to KNX may loose precision due to the KNX DPT 9 not being able to carry more than
+# two digits after the decimal point and having a mantissa of only 11 bits.
+# Consequently, when writing a 2-octet float to ebusd, a consecutive read on the same group address is likely to reveal
+# a different value if it was using more than two digits after the decimal point or exceeding the KNX float mantissa
+# range, e.g.:
+# - an ebus D2B value of 10.004 will read as 10.00 2-octet float on KNX,
+# - an ebus UIN with divisor 100 (like heating curve) value of 655.34 will read as 655.04 2-octet float on KNX,
+# - writing a KNX 2-octet float value of 12.34 to an ebus UIN with divisor 10 will actually write 12.3 and read as 12.3.
+#
+# note: writing to ebus via KNX currently is only possible if the ebus message contains a single field respectively at
+# most one non-ignored field. this is due to otherwise the value to be set for the other fields would have to be
+# determined first which is mostly not possible. Group associations to write messages not fulfilling this requirement
+# are silently ignored.
+#
+# note: the mapping for reads/writes from KNX is done as follows:
+# - for KNX read, the precedence on picking the ebus message is: active read, passive read+write.
+# - for KNX write, the precedence on picking the ebus message is: active write only.
+broadcast/datetime/outsidetemp = 9/10

src/ebusd/CMakeLists.txt

@@ -13,6 +13,11 @@ if(HAVE_MQTT)
   set(ebusd_LIBS ${ebusd_LIBS} mosquitto)
 endif(HAVE_MQTT)
 
+if(HAVE_KNX)
+  set(ebusd_SOURCES ${ebusd_SOURCES} knxhandler.cpp knxhandler.h)
+  set(ebusd_LIBS ${ebusd_LIBS} eibclient)
+endif(HAVE_KNX)
+
 if(HAVE_SSL)
   set(ebusd_LIBS ${ebusd_LIBS} ssl crypto)
 endif(HAVE_SSL)

src/ebusd/Makefile.am

@@ -16,6 +16,9 @@ ebusd_SOURCES = \
 if MQTT
 ebusd_SOURCES += mqtthandler.cpp mqtthandler.h
 endif
+if KNX
+ebusd_SOURCES += knxhandler.cpp knxhandler.h
+endif
 
 ebusd_LDADD = ../lib/utils/libutils.a \
               ../lib/ebus/libebus.a \

src/ebusd/datahandler.cpp

@@ -24,6 +24,9 @@
 #ifdef HAVE_MQTT
 #  include "ebusd/mqtthandler.h"
 #endif
+#ifdef HAVE_KNX
+#  include "ebusd/knxhandler.h"
+#endif
 
 namespace ebusd {
 
@@ -36,12 +39,18 @@ static struct argp_child g_argp_children[
 #ifdef HAVE_MQTT
             +1
 #endif
+#ifdef HAVE_KNX
+            +1
+#endif
 ];
 
 const struct argp_child* datahandler_getargs() {
   size_t count = 0;
 #ifdef HAVE_MQTT
   g_argp_children[count++] = *mqtthandler_getargs();
+#endif
+#ifdef HAVE_KNX
+  g_argp_children[count++] = *knxhandler_getargs();
 #endif
   if (count > 0) {
     g_argp_children[count] = g_last_argp_child;
@@ -57,6 +66,11 @@ bool datahandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap
   if (!mqtthandler_register(userInfo, busHandler, messages, handlers)) {
     success = false;
   }
+#endif
+#ifdef HAVE_KNX
+  if (!knxhandler_register(userInfo, busHandler, messages, handlers)) {
+    success = false;
+  }
 #endif
   return success;
 }

src/ebusd/knxhandler.cpp

@@ -0,0 +1,837 @@
+/*
+ * ebusd - daemon for communication with eBUS heating systems.
+ * Copyright (C) 2022 John Baier <ebusd@ebusd.eu>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifdef HAVE_CONFIG_H
+#  include <config.h>
+#endif
+
+#include "ebusd/knxhandler.h"
+#ifdef HAVE_PPOLL
+#  include <poll.h>
+#endif
+#include <cmath>
+#include <csignal>
+#include <deque>
+#include "lib/utils/log.h"
+#include "lib/ebus/symbol.h"
+
+namespace ebusd {
+
+using std::dec;
+
+// version is coded as:
+// 5 bits magic (to be incremented with incompatible changes, not shown)
+// 5 bits major, using major directly
+// 6 bits minor, using minor multiplied by 10 to have space for micro versioning in future
+#define VERSION_INT ((PACKAGE_VERSION_MAJOR<<6)|(PACKAGE_VERSION_MINOR*10))
+
+#define O_URL 1
+#define O_INT (O_URL+1)
+#define O_VAR (O_INT+1)
+
+/** the definition of the KNX arguments. */
+static const struct argp_option g_knx_argp_options[] = {
+  {nullptr,      0, nullptr,      0, "KNX options:", 1 },
+  {"knxurl", O_URL,   "URL",      0, "Connect to KNX daemon on URL (i.e. \"ip:host:[port]\" or \"local:/socketpath\") []", 0 },
+  {"knxint", O_INT,  "FILE",      0, "Read KNX integration settings from FILE [/etc/ebusd/knx.cfg]", 0 },
+  {"knxvar", O_VAR, "NAME=VALUE", 0, "Add a variable to the read KNX integration settings", 0 },
+
+  {nullptr,      0, nullptr,      0, nullptr, 0 },
+};
+
+static const char* g_url = nullptr;  //!< URL of KNX daemon
+static const char* g_integrationFile = nullptr;  //!< the integration settings file
+static vector<string>* g_integrationVars = nullptr;  //!< the integration settings variables
+
+/**
+ * The KNX argument parsing function.
+ * @param key the key from @a g_knx_argp_options.
+ * @param arg the option argument, or nullptr.
+ * @param state the parsing state.
+ */
+static error_t knx_parse_opt(int key, char *arg, struct argp_state *state) {
+  switch (key) {
+  case O_URL:  // --knxurl=localhost
+    if (arg == nullptr || arg[0] == 0) {
+      argp_error(state, "invalid knxurl");
+      return EINVAL;
+    }
+    g_url = arg;
+    break;
+
+  case O_INT:  // --knxint=/etc/ebusd/knx.cfg
+    if (arg == nullptr || arg[0] == 0 || strcmp("/", arg) == 0) {
+      argp_error(state, "invalid knxint file");
+      return EINVAL;
+    }
+    g_integrationFile = arg;
+    break;
+
+  case O_VAR:  // --knxvar=NAME=VALUE
+    if (arg == nullptr || arg[0] == 0 || !strchr(arg, '=')) {
+      argp_error(state, "invalid knxvar");
+      return EINVAL;
+    }
+    if (!g_integrationVars) {
+      g_integrationVars = new vector<string>();
+    }
+    g_integrationVars->push_back(string(arg));
+    break;
+
+  default:
+    return ARGP_ERR_UNKNOWN;
+  }
+  return 0;
+}
+
+static const struct argp g_knx_argp = { g_knx_argp_options, knx_parse_opt, nullptr, nullptr, nullptr, nullptr,
+  nullptr };
+static const struct argp_child g_knx_argp_child = {&g_knx_argp, 0, "", 1};
+
+
+const struct argp_child* knxhandler_getargs() {
+  return &g_knx_argp_child;
+}
+
+bool knxhandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages,
+    list<DataHandler*>* handlers) {
+  if (g_url) {
+    handlers->push_back(new KnxHandler(userInfo, busHandler, messages));
+  }
+  return true;
+}
+
+KnxHandler::KnxHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages)
+  : DataSink(userInfo, "knx"), DataSource(busHandler), WaitThread(), m_messages(messages),
+    m_start(0), m_con(nullptr), m_lastUpdateCheckResult("."),
+    m_lastScanStatus(SCAN_STATUS_NONE), m_scanFinishReceived(false), m_lastErrorLogTime(0) {
+  if (g_integrationFile != nullptr) {
+    if (!m_replacers.parseFile(g_integrationFile)) {
+      logOtherError("knx", "unable to open integration file %s", g_integrationFile);
+    }
+  }
+  if (g_integrationVars) {
+    for (auto& str : *g_integrationVars) {
+      m_replacers.parseLine(str);
+    }
+    delete g_integrationVars;
+    g_integrationVars = nullptr;
+  }
+  // parse all group to message field assignments
+  vector<string> keys = m_replacers.keys();
+  for (auto& key : keys) {
+    auto pos = key.find('/');
+    if (pos == string::npos) {
+      continue;
+    }
+    string val = m_replacers.get(key, false);
+    pos = val.find('/');
+    if (pos == string::npos) {
+      continue;
+    }
+    auto pos2 = val.find('/', pos+1);
+    result_t res = RESULT_OK;
+    unsigned int v;
+    v = parseInt(val.substr(0, pos).c_str(), 10, 0, 0x1f, &res);
+    if (res != RESULT_OK) {
+      continue;
+    }
+    auto dest = static_cast<eibaddr_t>(v << 11);
+    if (pos2 == string::npos) {
+      // 2 level
+      v = parseInt(val.substr(pos+1).c_str(), 10, 0, 0x7ff, &res);
+      if (res != RESULT_OK) {
+        continue;
+      }
+      dest |= static_cast<eibaddr_t>(v);
+    } else {
+      // 3 level
+      v = parseInt(val.substr(pos+1, pos2).c_str(), 10, 0, 0x07, &res);
+      if (res != RESULT_OK) {
+        continue;
+      }
+      dest |= static_cast<eibaddr_t>(v << 8);
+      v = parseInt(val.substr(pos+1, pos2).c_str(), 10, 0, 0xff, &res);
+      if (res != RESULT_OK) {
+        continue;
+      }
+      dest |= static_cast<eibaddr_t>(v);
+    }
+    if (key.substr(0, 7) != "global/") {
+      m_messageFieldGroupAddress[key] = dest;
+      continue;
+    }
+    key = key.substr(7);
+    global_t index;
+    dtlf_t lengthFlag = DTLF_1BIT;  // default for <=6 bits
+    if (key == "version") {
+      index = GLOBAL_VERSION;
+      lengthFlag.length = 2;
+    } else if (key == "running") {
+      index = GLOBAL_RUNNING;
+    } else if (key == "uptime") {
+      index = GLOBAL_UPTIME;
+      lengthFlag.length = 4;
+    } else if (key == "signal") {
+      index = GLOBAL_SIGNAL;
+    } else if (key == "scan") {
+      index = GLOBAL_SCAN;
+    } else if (key == "updatecheck") {
+      index = GLOBAL_UPDATECHECK;
+    } else {
+      continue;
+    }
+    m_subscribedGlobals[index] = dest|FLAG_READ;
+    m_subscribedGroups[dest|FLAG_READ] = {
+        .messageKey = 0,
+        .globalIndex = index,
+        .lengthFlag = lengthFlag,
+    };
+  }
+}
+
+KnxHandler::~KnxHandler() {
+  join();
+  if (m_con) {
+    EIBClose(m_con);
+    m_con = nullptr;
+  }
+}
+
+void KnxHandler::startHandler() {
+  WaitThread::start("KNX");
+}
+
+void KnxHandler::notifyUpdateCheckResult(const string& checkResult) {
+  if (checkResult != m_lastUpdateCheckResult) {
+    m_lastUpdateCheckResult = checkResult;
+    sendGlobalValue(GLOBAL_UPDATECHECK, checkResult.empty() || checkResult=="OK" ? 0 : 1);
+  }
+}
+
+void KnxHandler::notifyScanStatus(scanStatus_t scanStatus) {
+  if (scanStatus == SCAN_STATUS_FINISHED) {
+    m_scanFinishReceived = true;
+  }
+  if (scanStatus != m_lastScanStatus) {
+    m_lastScanStatus = scanStatus;
+    sendGlobalValue(GLOBAL_SCAN, m_lastScanStatus==SCAN_STATUS_RUNNING ? 1 : 0);
+  }
+}
+
+result_t getFieldLength(const SingleDataField *field, dtlf_t *length) {
+  const auto dt = field->getDataType();
+  if (field->isIgnored() || !dt->isNumeric() || dt->isAdjustableLength()) {
+    return RESULT_ERR_INVALID_NUM;
+  }
+  size_t bitCnt = dt->getBitCount();
+  if (bitCnt == 1) {
+    *length = DTLF_1BIT;
+    return RESULT_OK;
+  }
+  if (bitCnt < 8) {
+    *length = DTLF_8BIT;
+    return RESULT_OK;
+  }
+  const auto nt = dynamic_cast<const NumberDataType*>(dt);
+  if (nt->getDivisor()!=1) {
+    // adjust bit count to 2 octet or 4 octet float DPT
+    if (bitCnt>=24 && bitCnt<31) {
+      bitCnt = 32;
+    } else if (bitCnt<16) {
+      bitCnt = 16;
+    }
+    // TODO uncommon divisor (e.g. >100) may not fit into KNX 2-octet float or truncates precision
+  } else if (bitCnt>=24 && bitCnt<31) {
+    // adjust bit count for non-existent 24 bit KNX type
+    bitCnt = 32;
+  }
+  *length = {{
+    .hasDivisor = nt->getDivisor()!=1,
+    .isFloat = dt->hasFlag(EXP),
+    .isSigned = dt->hasFlag(SIG),
+    .length = static_cast<uint8_t>(bitCnt/8),
+  }};
+  return RESULT_OK;
+}
+
+uint32_t floatToInt16(float val) {
+  // (0.01*m)(2^e) format with sign, 12 bits mantissa (incl. sign), 4 bits exponent
+  if (val == 0) {
+    return 0;
+  }
+  bool negative = val < 0;
+  if (negative) {
+    val = -val;
+  }
+  val *= 100;
+  int exp = ilogb(val)-10;
+  if (exp < -10 || exp > 15) {
+    return 0x7fff;  // invalid value DPT 9
+  }
+  auto shift = exp > 0 ? exp : 0;
+  auto sig = static_cast<uint32_t>(val * exp2(-shift));
+  uint32_t value = static_cast<uint32_t>(shift << 11) | sig;
+  if (negative) {
+    return value | 0x8000;
+  }
+  return value;
+}
+
+float int16ToFloat(uint16_t val) {
+  if (val == 0) {
+    return 0;
+  }
+  if (val == 0x7fff) {
+    return static_cast<float>(0xffffffff); // NaN
+  }
+  bool negative = val&0x8000;
+  int exp = (val>>11)&0xf;
+  int sig = val&0x7ff;
+  return static_cast<float>(sig * exp2(exp) * (negative ? -0.01 : 0.01));
+}
+
+result_t KnxHandler::sendGroupValue(eibaddr_t dest, apci_t apci, dtlf_t& lengthFlag, unsigned int value, const SingleDataField *field) const {
+  uint8_t data[] = {0, 0, 0, 0, 0, 0};
+  data[0] = static_cast<uint8_t>(apci>>8);
+  data[1] = static_cast<uint8_t>(apci&0xff);
+  int len = 2;
+  // convert value to dpt
+  if (lengthFlag.isFloat || lengthFlag.hasDivisor) {
+    if (!field) {
+      return RESULT_ERR_INVALID_NUM;
+    }
+    auto nt = dynamic_cast<const NumberDataType*>(field->getDataType());
+    float fval;
+    result_t ret = nt->getFloatFromRawValue(value, &fval);
+    if (ret == RESULT_EMPTY) {
+      // replacement value:
+      if (lengthFlag.length==2) {
+        // shall have 0x7fff for DPT 9
+        value = 0x7fff;
+      } else {
+        return RESULT_ERR_INVALID_NUM;  // not encodable
+      }
+    } else if (ret != RESULT_OK) {
+      return ret;
+    } else if (lengthFlag.length == 2) {
+      // convert to (0.01*m)(2^e) format with sign, 12 bits mantissa (incl. sign), 4 bits exponent
+      value = floatToInt16(fval);
+    } else if (lengthFlag.length == 4) {
+      // convert to IEEE 754
+      value = floatToUint(fval);
+    } else {
+      return RESULT_ERR_INVALID_NUM;  // not encodable
+    }
+  }
+  // else signed values: fine as long as length is identical
+  if (apci==APCI_GROUPVALUE_WRITE && lengthFlag.lastValueSent && lengthFlag.lastValue==value) {
+    return RESULT_EMPTY;  // no need to send the same group value again
+  }
+  lengthFlag.lastValue = value;
+  lengthFlag.lastValueSent = true;
+  switch (lengthFlag.length) {
+    case 0:  // short value <= 6 bit
+      data[1] |= static_cast<uint8_t>(value&0x3f);
+      break;
+    case 1:  // 1 octet
+      data[2] = static_cast<uint8_t>(value&0xff);
+      break;
+    case 2:  // 2 octets
+      data[2] = static_cast<uint8_t>(value>>8);
+      data[3] = static_cast<uint8_t>(value&0xff);
+      break;
+    case 4:  // 4 octets
+      data[2] = static_cast<uint8_t>(value>>24);
+      data[3] = static_cast<uint8_t>(value>>16);
+      data[4] = static_cast<uint8_t>(value>>8);
+      data[5] = static_cast<uint8_t>(value&0xff);
+      break;
+    default:
+      return RESULT_ERR_INVALID_NUM;
+  }
+  len += lengthFlag.length;
+  if (EIBSendGroup(m_con, dest, len, data) < 0) {
+    return RESULT_ERR_SEND;
+  }
+  return RESULT_OK;
+}
+
+void KnxHandler::sendGlobalValue(global_t index, unsigned int value, bool response) {
+  if (!m_con) {
+    return;
+  }
+  const auto vit = m_subscribedGlobals.find(index);
+  if (vit == m_subscribedGlobals.cend()) {
+    return;
+  }
+  auto git = m_subscribedGroups.find(vit->second);
+  if (git == m_subscribedGroups.end()) {
+    return;
+  }
+  sendGroupValue(static_cast<eibaddr_t>(vit->second&0xffff),
+                 response ? APCI_GROUPVALUE_RESPONSE : APCI_GROUPVALUE_WRITE,
+                 git->second.lengthFlag, value);
+}
+
+result_t KnxHandler::receiveTelegram(int maxlen, uint8_t *buf, int *recvlen,
+                                  eibaddr_t *src, eibaddr_t *dest) {
+  struct timespec tdiff = {
+      .tv_sec = 2,
+      .tv_nsec = 0,
+  };
+  int fd = EIB_Poll_FD(m_con);
+#ifdef HAVE_PPOLL
+  nfds_t nfds = 1;
+  struct pollfd fds[nfds];
+  memset(fds, 0, sizeof(fds));
+  fds[0].fd = fd;
+  fds[0].events = POLLIN | POLLERR | POLLHUP | POLLRDHUP;
+#else
+#ifdef HAVE_PSELECT
+  fd_set checkfds, exceptfds;
+  FD_ZERO(&checkfds);
+  FD_SET(fd, &checkfds);
+  FD_ZERO(&exceptfds);
+  FD_SET(fd, &exceptfds);
+#endif
+#endif
+  int ret;
+#ifdef HAVE_PPOLL
+  ret = ppoll(fds, nfds, &tdiff, nullptr);
+#else
+#ifdef HAVE_PSELECT
+  fd_set readfds = checkfds;
+  ret = pselect(fd + 1, &readfds, nullptr, &exceptfds, &tdiff, nullptr);
+#endif
+#endif
+  bool newData;
+#ifdef HAVE_PPOLL
+  if (ret < 0 || (ret > 0 && (fds[0].revents & (POLLERR | POLLHUP | POLLRDHUP)))) {
+    return RESULT_ERR_GENERIC_IO;
+  }
+  newData = fds[0].revents & POLLIN;
+#else
+#ifdef HAVE_PSELECT
+  if (ret < 0 || FD_ISSET(fd, &exceptfds)) {
+    return RESULT_ERR_GENERIC_IO;
+  }
+  newData = FD_ISSET(fd, &readfds);
+#endif
+#endif
+  int len = EIB_Poll_Complete(m_con);
+  if (len == -1) {
+    // read failed
+    return RESULT_ERR_GENERIC_IO;
+  }
+  if (!newData) {
+    // timeout
+    return RESULT_ERR_TIMEOUT;
+  }
+  len = EIBGetGroup_Src(m_con, maxlen, buf, src, dest);
+  if (len < 0) {
+    return RESULT_ERR_GENERIC_IO;
+  }
+  if (len < 2) {
+    return RESULT_ERR_GENERIC_IO;
+  }
+  *recvlen = len;
+  return RESULT_OK;
+}
+
+void printResponse(eibaddr_t src, eibaddr_t dest, int len, const uint8_t *data) {
+  int apci = ((data[0]&0x03)<<2) | ((data[1]&0xc0)>>6);
+  int value = len==2 ? data[1]&0x3f : data[2]; // 6 bits or full octet
+  if (len>3) {
+    value = (value<<8) | data[3]; // up to 16 bits
+  }
+  if (len>4) {
+    value = (value<<8) | data[4]; // up to 24 bits
+  }
+  if (len>5) {
+    value = (value<<8) | data[5]; // up to 32 bits
+  }
+  logOtherDebug("knx", "recv from %4.4x to %4.4x, %d=%s, len %d, value %d", src, dest,
+                apci, apci==0?"read":apci==2?"write":apci==1?"resp":"other", len, value);
+}
+
+void KnxHandler::handleReceivedTelegram(eibaddr_t src, eibaddr_t dest, int len, const uint8_t *data) {
+  int apci = ((data[0]&0x03)<<8) | data[1];
+  if ((apci & APCI_GROUPVALUE_READ_MASK) == 0) {
+    apci &= ~APCI_GROUPVALUE_READ_MASK;
+  }
+  bool isWrite = apci==APCI_GROUPVALUE_WRITE;
+  if (apci!=APCI_GROUPVALUE_READ && !isWrite) {
+    return;  // neither A_GroupValue_Read nor A_GroupValue_Write (A_GroupValue_Response not used at all)
+  }
+  const auto subKey = static_cast<uint32_t>(dest | (isWrite ? FLAG_WRITE : FLAG_READ));
+  auto sit = m_subscribedGroups.find(subKey);
+  if (sit == m_subscribedGroups.end()) {
+    return;  // address+direction not subscribed
+  }
+  if (sit->second.messageKey == 0) {
+    // global values, only readable
+    switch (sit->second.globalIndex) {
+      case GLOBAL_VERSION:
+        sendGlobalValue(GLOBAL_VERSION, VERSION_INT, true);
+        break;
+      case GLOBAL_RUNNING:
+        sendGlobalValue(GLOBAL_RUNNING, 1, true);
+        break;
+      case GLOBAL_UPTIME:
+        sendGlobalValue(GLOBAL_UPTIME, static_cast<unsigned>(time(nullptr) - m_start), true);
+        break;
+      case GLOBAL_SIGNAL:
+        sendGlobalValue(GLOBAL_SIGNAL, m_busHandler->hasSignal() ? 1 : 0, true);
+        break;
+      case GLOBAL_SCAN:
+        sendGlobalValue(GLOBAL_SCAN, m_lastScanStatus==SCAN_STATUS_RUNNING ? 1 : 0, true);
+        break;
+      case GLOBAL_UPDATECHECK:
+        sendGlobalValue(GLOBAL_UPDATECHECK, m_lastUpdateCheckResult.empty() || m_lastUpdateCheckResult=="OK" || m_lastUpdateCheckResult=="." ? 0 : 1, true);
+        break;
+      default:
+        return;  // ignore
+    }
+    return;
+  }
+  const vector<Message*>* messages = m_messages->getByKey(sit->second.messageKey);
+  if (!messages) {
+    return;
+  }
+  Message *msg = nullptr;
+  ssize_t fieldIndex = sit->second.fieldIndex;
+  const SingleDataField* field = nullptr;
+  for (const auto& message : *messages) {
+    if (!message->isAvailable() || message->getDstAddress() == SYN) {
+      continue;
+    }
+    if ((message->isWrite() && !message->isPassive()) != isWrite) {
+      continue;
+    }
+    field = message->getField(fieldIndex);
+    if (!field) {
+      continue;
+    }
+    if (isWrite) {
+      msg = message;
+      break;  // best candidate
+    }
+    if (!msg) {
+      msg = message;
+    } else if (message->getLastUpdateTime() > 0
+    && message->getLastUpdateTime() > msg->getLastUpdateTime()) {
+      // prefer newer updated, even if it is passive
+      msg = message;
+    } else if (!message->isPassive()) {
+      // prefer active read before passive
+      msg = message;
+    }
+  }
+  if (!msg) {
+    return;
+  }
+  result_t res;
+  const string circuit = msg->getCircuit(), name = msg->getName(), fieldName = msg->getFieldName(fieldIndex);
+  if (isWrite) {
+    unsigned int value = len==2 ? data[1]&0x3f : data[2]; // <=6 bits or full octet
+    if (len>3) {
+      value = (value<<8) | data[3]; // up to 16 bits
+    }
+    if (len>4) {
+      value = (value<<8) | data[4]; // up to 24 bits
+    }
+    if (len>5) {
+      value = (value<<8) | data[5]; // up to 32 bits
+    }
+    // TODO write from KNX updates the message and thus re-sends the write later on again
+    logOtherNotice("knx", "received write request from %4.4x to %4.4x for %s/%s/%s, value %d",
+                   src, dest, circuit.c_str(), name.c_str(), fieldName.c_str(), value);
+    // write new field value to bus if possible
+    // ugly but least intrusive: format single num field value to string to have it parsed back later on
+    ostringstream str;
+    // convert value to dpt
+    auto lengthFlag = sit->second.lengthFlag;
+    if (lengthFlag.isFloat || lengthFlag.hasDivisor) {
+      float fval;
+      if (lengthFlag.length == 2) {
+        // convert from (0.01*m)(2^e) format with sign, 12 bits mantissa (incl. sign), 4 bits exponent
+        fval = int16ToFloat(static_cast<uint16_t>(value));
+      } else if (lengthFlag.length == 4) {
+        // convert from IEEE 754
+        fval = uintToFloat(value);
+      } else {
+        return;  // not decodable
+      }
+      str << static_cast<float>(fval);
+    } else {
+      if (lengthFlag.isSigned) {
+        // signed values: determine sign
+        uint32_t bit = 1<<(lengthFlag.length*8-1);
+        if (value & bit) {
+          value = -(value&~bit);
+        }
+        str << static_cast<int>(value);
+      } else {
+        str << static_cast<uint32_t>(value);
+      }
+    }
+    res = m_busHandler->readFromBus(msg, str.str());
+    if (res != RESULT_OK) {
+      logOtherError("knx", "write %s %s: %s", circuit.c_str(), name.c_str(), getResultCode(res));
+    }
+    return;
+  }
+  logOtherNotice("knx", "received read request from %4.4x to %4.4x for %s/%s/%s",
+                 src, dest, circuit.c_str(), name.c_str(), fieldName.c_str());
+  if (msg->getLastUpdateTime() <= 0) {  // TODO adjustable max age
+    res = m_busHandler->readFromBus(msg, "");
+    if (res != RESULT_OK) {
+      return;
+    }
+  }
+  unsigned int value = 0;
+  res = msg->decodeLastDataNumField(nullptr, fieldIndex, &value);
+  if (res == RESULT_OK) {
+    res = sendGroupValue(dest, APCI_GROUPVALUE_RESPONSE, sit->second.lengthFlag, value, field);
+  }
+}
+
+// interval in seconds for sending the uptime value
+#define UPTIME_INTERVAL 3600
+
+void KnxHandler::run() {
+  time_t lastTaskRun, now, lastSignal = 0, lastUptime = 0, lastUpdates = 0;
+  bool signal = false;
+  result_t result = RESULT_OK;
+  time(&now);
+  m_start = lastTaskRun = now;
+  uint8_t data[] = {0, 0, 0, 0, 0, 0, 0, 0};
+  int len = 0;
+  time_t definitionsSince = 0;
+  while (isRunning()) {
+    bool wasConnected = m_con != nullptr;
+    bool needsWait = true;
+    if (!m_con) {
+      m_con = EIBSocketURL(g_url);
+      const char* err = nullptr;
+      if (!m_con) {
+        err = "open error";
+      } else if (EIBOpen_GroupSocket(m_con, 0) < 0) {
+        err = "open group error";
+        EIBClose_sync(m_con);
+        m_con = nullptr;
+      } else {
+        m_lastErrorLogTime = 0;
+        logOtherNotice("knx", "connected");
+        sendGlobalValue(GLOBAL_VERSION, VERSION_INT);
+        sendGlobalValue(GLOBAL_RUNNING, 1);
+      }
+      if (err) {
+        time(&now);
+        if (now > m_lastErrorLogTime + 10) {  // log at most every 10 seconds
+          m_lastErrorLogTime = now;
+          logOtherError("knx", err);
+        }
+      }
+    }
+    bool reconnected = !wasConnected && m_con != nullptr;
+    time(&now);
+    bool sendSignal = reconnected;
+    if (now < m_start) {
+      // clock skew
+      if (now < lastSignal) {
+        lastSignal -= lastTaskRun-now;
+      }
+      lastTaskRun = now;
+    } else if (now > lastTaskRun+(m_scanFinishReceived ? 1 : 15)) {
+      m_scanFinishReceived = false;
+      if (m_con) {
+        sendSignal = true;
+        if (now > lastUptime + UPTIME_INTERVAL) {
+          lastUptime = now;
+          sendGlobalValue(GLOBAL_UPTIME, static_cast<unsigned int>(now - m_start));
+        }
+      }
+      if (m_con && definitionsSince == 0) {
+        definitionsSince = 1;
+      }
+      if (m_con) {
+        deque<Message*> messages;
+        m_messages->findAll("", "", m_levels, false, true, true, true, true, true, 0, 0, true, &messages);
+        for (const auto& message : messages) {
+          const auto mit = m_subscribedMessages.find(message->getKey());
+          if (mit != m_subscribedMessages.cend()) {
+            continue;  // already subscribed
+          }
+          if (message->getDstAddress() == SYN) {
+            continue;
+          }
+          bool isWrite = message->isWrite() && !message->isPassive();  // from KNX perspective
+          if (message->getCreateTime() <= definitionsSince) {  // only newer defined
+            continue;
+          }
+          ssize_t fieldCount = static_cast<signed>(message->getFieldCount());
+          if (isWrite && fieldCount>1) {
+            // impossible with more than one field
+            continue;
+          }
+          bool added = false;
+          for (ssize_t index = 0; index < fieldCount; index++) {
+            const SingleDataField* field = message->getField(index);
+            if (!field || field->isIgnored()) {
+              continue;
+            }
+            string fieldName = message->getFieldName(index);
+            if (fieldName.empty() && fieldCount == 1) {
+              fieldName = "0";  // might occur for unnamed single field sets
+            }
+            string key = message->getCircuit()+"/"+message->getName()+"/"+fieldName;
+            const auto git = m_messageFieldGroupAddress.find(key);
+            if (git == m_messageFieldGroupAddress.cend()) {
+              continue;
+            }
+            // determine field length in telegram
+            dtlf_t lengthFlag = {};
+            result = getFieldLength(field, &lengthFlag);
+            if (result != RESULT_OK) {
+              continue;
+            }
+            // store association
+            // TODO add "foreign" associations as well, i.e. read for a write msg and write for read msg?
+            eibaddr_t dest = git->second;
+            auto subKey = static_cast<uint32_t>(dest | (isWrite ? FLAG_WRITE : FLAG_READ));
+            auto sit = m_subscribedGroups.find(subKey);
+            if (sit != m_subscribedGroups.cend()) {
+              continue;
+            }
+            m_subscribedGroups[subKey] = {
+                .messageKey = message->getKey(),
+                .fieldIndex = static_cast<uint8_t>(index),
+                .lengthFlag = lengthFlag,
+            };
+            m_subscribedMessages[message->getKey()].push_back(subKey);
+            added = true;
+          }
+          if (!added) {
+            continue;
+          }
+          if (message->getLastUpdateTime() > message->getCreateTime()) {
+            // ensure data is published as well
+            m_updatedMessages[message->getKey()]++;
+          } else if (message->isWrite()) {
+            // publish data for read pendant of write message
+            Message* read = m_messages->find(message->getCircuit(), message->getName(), "", false);
+            if (read && read->getLastUpdateTime() > 0) {
+              m_updatedMessages[read->getKey()]++;
+            }
+          }
+        }
+        definitionsSince = now;
+        needsWait = true;
+      }
+      time(&lastTaskRun);
+    }
+    if (sendSignal) {
+      if (m_busHandler->hasSignal()) {
+        lastSignal = now;
+        if (!signal || reconnected) {
+          signal = true;
+          sendGlobalValue(GLOBAL_SIGNAL, 1);
+        }
+      } else {
+        if (signal || reconnected) {
+          signal = false;
+          sendGlobalValue(GLOBAL_SIGNAL, 0);
+        }
+      }
+    }
+    if (m_con) {
+      eibaddr_t src, dest;
+      // APDU data starting with octet 6 according to spec, contains 2 bits of application layer
+      result_t res = RESULT_OK;
+      do {
+        res = receiveTelegram(8, data, &len, &src, &dest);
+        if (res != RESULT_OK) {
+          if (res == RESULT_ERR_GENERIC_IO) {
+            EIBClose_sync(m_con);
+            m_con = nullptr;
+          }
+        } else {
+          needsWait = false;
+          printResponse(src, dest, len, data);
+          handleReceivedTelegram(src, dest, len, data);
+        }
+      } while (res == RESULT_OK);
+    }
+    if (!m_updatedMessages.empty()) {
+      m_messages->lock();
+      if (m_con) {
+        for (auto it = m_updatedMessages.begin(); it != m_updatedMessages.end(); ) {
+          const vector<Message*>* messages = m_messages->getByKey(it->first);
+          if (!messages) {
+            continue;
+          }
+          for (const auto& message : *messages) {
+            if (message->getLastChangeTime() <= 0) {
+              continue;
+            }
+            const auto mit = m_subscribedMessages.find(message->getKey());
+            if (mit == m_subscribedMessages.cend()) {
+              continue;
+            }
+            if (!(message->getDataHandlerState()&2)) {
+              message->setDataHandlerState(2, true);  // first update still needed
+            } else if (message->getLastChangeTime() <= lastUpdates) {
+              continue;
+            }
+            for (auto destFlags : mit->second) {
+              bool isWrite = (destFlags&FLAG_WRITE)!=0;  // from KNX perspective
+              auto sit = m_subscribedGroups.find(destFlags);
+              if (sit == m_subscribedGroups.end()) {
+                continue;
+              }
+              ssize_t index = sit->second.fieldIndex;
+              const SingleDataField *field = message->getField(index);
+              if (!field || field->isIgnored()) {
+                continue;
+              }
+              eibaddr_t dest = destFlags&0xffff;
+              unsigned int value = 0;
+              result = message->decodeLastDataNumField(nullptr, index, &value);
+              sendGroupValue(dest, APCI_GROUPVALUE_WRITE, sit->second.lengthFlag, value, field);
+            }
+          }
+          it = m_updatedMessages.erase(it);
+        }
+        time(&lastUpdates);
+      } else {
+        m_updatedMessages.clear();
+      }
+      m_messages->unlock();
+    }
+    if ((!m_con && !Wait(5)) || (needsWait && !Wait(1))) {
+      break;
+    }
+  }
+  sendGlobalValue(GLOBAL_RUNNING, 0);
+  sendGlobalValue(GLOBAL_SIGNAL, 0);
+  sendGlobalValue(GLOBAL_SCAN, 0);
+}
+
+}  // namespace ebusd

src/ebusd/knxhandler.h

@@ -0,0 +1,230 @@
+/*
+ * ebusd - daemon for communication with eBUS heating systems.
+ * Copyright (C) 2022 John Baier <ebusd@ebusd.eu>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef EBUSD_KNXHANDLER_H_
+#define EBUSD_KNXHANDLER_H_
+
+#include <eibclient.h>
+#include <map>
+#include <string>
+#include <list>
+#include <vector>
+#include <utility>
+#include "ebusd/datahandler.h"
+#include "ebusd/bushandler.h"
+#include "lib/ebus/message.h"
+#include "lib/ebus/stringhelper.h"
+
+namespace ebusd {
+
+/** @file ebusd/knxhandler.h
+ * A data handler enabling KNX support via knx.
+ */
+
+using std::map;
+using std::string;
+using std::vector;
+
+/**
+ * Helper function for getting the argp definition for KNX.
+ * @return a pointer to the argp_child structure.
+ */
+const struct argp_child* knxhandler_getargs();
+
+/**
+ * Registration function that is called once during initialization.
+ * @param userInfo the @a UserInfo instance.
+ * @param busHandler the @a BusHandler instance.
+ * @param messages the @a MessageMap instance.
+ * @param handlers the @a list to which new @a DataHandler instances shall be added.
+ * @return true if registration was successful.
+ */
+bool knxhandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages,
+    list<DataHandler*>* handlers);
+
+
+/** type for KNX APCI values (application control field). */
+enum apci_t {
+  APCI_GROUPVALUE_READ = 0x000,      //!< A_GroupValue_Read-PDU
+  APCI_GROUPVALUE_RESPONSE = 0x040,  //!< A_GroupValue_Response-PDU
+  APCI_GROUPVALUE_WRITE = 0x080,     //!< A_GroupValue_Write-PDU
+};
+
+#define APCI_GROUPVALUE_READ_MASK 0x3c0
+
+#define FLAG_READ 0x400000
+#define FLAG_WRITE 0x800000
+
+/** datatype length flags (byte length on KNX in bits 0-3, extra info in higher bits). */
+typedef union {
+  struct {
+    bool hasDivisor: 1;
+    bool isFloat: 1;
+    bool isSigned: 1;
+    bool lastValueSent: 1;
+    uint8_t length; // 0 for 1-6 bits, number of bytes otherwise
+    uint32_t lastValue;
+  };
+  uint64_t value;
+} dtlf_t;
+
+#define DTLF_1BIT {.length = 0}
+#define DTLF_8BIT {.length = 1}
+
+/** type for global values not associated with an ebus message. */
+enum global_t {
+  GLOBAL_VERSION = 1,
+  GLOBAL_RUNNING = 2,
+  GLOBAL_UPTIME = 3,
+  GLOBAL_SIGNAL = 4,
+  GLOBAL_SCAN = 5,
+  GLOBAL_UPDATECHECK = 6,
+};
+
+/** type for several group subscription infos. */
+typedef struct {
+  uint64_t messageKey;  // message key, or 0 for global value
+  union {
+    uint8_t fieldIndex;  // message field index
+    global_t globalIndex;  // global value index
+  };
+  dtlf_t lengthFlag;  // telegram length and flags
+} groupInfo_t;
+
+
+/**
+ * The main class supporting KNX data handling.
+ */
+class KnxHandler : public DataSink, public DataSource, public WaitThread {
+ public:
+  /**
+   * Constructor.
+   * @param userInfo the @a UserInfo instance.
+   * @param busHandler the @a BusHandler instance.
+   * @param messages the @a MessageMap instance.
+   */
+  KnxHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages);
+
+ public:
+  /**
+   * Destructor.
+   */
+  ~KnxHandler() override;
+
+  // @copydoc
+  void startHandler() override;
+
+  // @copydoc
+  void notifyUpdateCheckResult(const string& checkResult) override;
+
+  // @copydoc
+  void notifyScanStatus(scanStatus_t scanStatus) override;
+
+  /**
+   * Send a group value.
+   * @param dest the destination group address.
+   * @param apci the APCI value.
+   * @param lengthFlag the datatype length flag.
+   * @param value the value.
+   * @param field the message field or nullptr for non field related.
+   * @return the result code.
+   */
+  result_t sendGroupValue(eibaddr_t dest, apci_t apci, dtlf_t& lengthFlag, unsigned int value, const SingleDataField *field = nullptr) const;
+
+  /**
+   * Send a global value to the registered group address.
+   * @param index the global value index to send.
+   * @param value the raw value.
+   * @param response true to send as response, false to send as write.
+   */
+  void sendGlobalValue(global_t index, unsigned int value, bool response = false);
+
+  /**
+   * Wait for and receive a KNX group telegram.
+   * @param maxlen the size of the data buffer.
+   * @param buf the data buffer.
+   * @param recvlen pointer to a variable in which to store the actually received length.
+   * @param src pointer to a variable in which to store the source address.
+   * @param dest pointer to a variable in which to store the destination group address.
+   * @return the result code, either RESULT_OK on success, RESULT_ERR_GENERIC_IO on I/O error (e.g. socket closed),
+   * or RESULT_ERR_TIMEOUT if no data is available.
+   */
+  result_t receiveTelegram(int maxlen, uint8_t *buf, int *recvlen, eibaddr_t *src, eibaddr_t *dest);
+
+  /**
+   * Handle a received KNX telegram.
+   * @param src the source address.
+   * @param dest the destination group address.
+   * @param len the telegram length (starting with ovctet 6).
+   * @param data the telegram data buffer.
+   */
+  void handleReceivedTelegram(eibaddr_t src, eibaddr_t dest, int len, const uint8_t *data);
+
+ protected:
+  // @copydoc
+  void run() override;
+
+
+ private:
+  /** the @a MessageMap instance. */
+  MessageMap* m_messages;
+
+  /** the @a StringReplacers from the integration file. */
+  StringReplacers m_replacers;
+
+  /** the group address for relevant message fields before being subscribed to by "circuit/message/field" name. */
+  map<string, eibaddr_t> m_messageFieldGroupAddress;
+
+  /**
+   * the group addresses that need to be responded to.
+   * key is the group address in lower 16 bits, and flags in upper 16 bits with:
+   * - read direction in bit 6 (<<16),
+   * - write direction in bit 7 (<<16).
+   * this way read and write may be mapped to different messages.
+   * value contains the message key and additional infos.
+   */
+  map<uint32_t, groupInfo_t>m_subscribedGroups;
+
+  /** the group address and flags (key of m_subscribedGroups) by subscribed message key. */
+  map<uint64_t, list<uint32_t>>m_subscribedMessages;
+
+  /** the group address and flags (key of m_subscribedGroups) by subscribed global values. */
+  map<global_t, uint32_t>m_subscribedGlobals;
+
+  /** the time the run thread was entered. */
+  time_t m_start;
+
+  /** the knx structure if initialized, or nullptr. */
+  EIBConnection* m_con;
+
+  /** the last update check result. */
+  string m_lastUpdateCheckResult;
+
+  /** the last scan status. */
+  scanStatus_t m_lastScanStatus;
+
+  /** set to true when a scan finish was received. */
+  bool m_scanFinishReceived;
+
+  /** the last system time when a communication error was logged. */
+  time_t m_lastErrorLogTime;
+};
+
+}  // namespace ebusd
+
+#endif  // EBUSD_KNXHANDLER_H_

src/ebusd/mainloop.cpp

@@ -327,6 +327,10 @@ void MainLoop::run() {
         if (m_messages->sizeConditions() > 0 && !m_polling) {
           logError(lf_main, "conditions require a poll interval > 0");
         }
+        // notify data sinks to make them update the messages
+        for (const auto dataSink : dataSinks) {
+          dataSink->notifyScanStatus(SCAN_STATUS_FINISHED);
+        }
       }
       if (m_runUpdateCheck && !m_shutdown && now > nextCheckRun) {
         if (!m_httpClient.connect("upd.ebusd.eu",

src/lib/ebus/datatype.cpp

@@ -793,6 +793,70 @@ result_t NumberDataType::readSymbols(size_t offset, size_t length, const SymbolS
   return readFromRawValue(value, outputFormat, output);
 }
 
+result_t NumberDataType::getFloatFromRawValue(unsigned int value, float* output) const {
+  int signedValue;
+  if (!hasFlag(REQ) && value == m_replacement) {
+    return RESULT_EMPTY;
+  }
+
+  bool negative;
+  if (hasFlag(SIG)) {  // signed value
+    negative = (value & (1 << (m_bitCount - 1))) != 0;
+    if (negative) {  // negative signed value
+      if (value < m_minValue) {
+        return RESULT_ERR_OUT_OF_RANGE;  // value out of range
+      }
+    } else if (value > m_maxValue) {
+      return RESULT_ERR_OUT_OF_RANGE;  // value out of range
+    }
+  } else if (value < m_minValue || value > m_maxValue) {
+    return RESULT_ERR_OUT_OF_RANGE;  // value out of range
+  } else {
+    negative = false;
+  }
+  if (m_bitCount == 32) {
+    if (hasFlag(EXP)) {  // IEEE 754 binary32
+      float val = uintToFloat(value);
+      if (val != val) {  // !isnan(val)
+        return RESULT_EMPTY;
+      }
+      if (val != 0.0) {
+        if (m_divisor < 0) {
+          val *= static_cast<float>(-m_divisor);
+        } else if (m_divisor > 1) {
+          val /= static_cast<float>(m_divisor);
+        }
+      }
+      *output = static_cast<float>(val);
+      return RESULT_OK;
+    }
+    // less than 32 bit
+    if (!negative) {
+      if (m_divisor < 0) {
+        *output = static_cast<float>(value) * static_cast<float>(-m_divisor);
+      } else if (m_divisor <= 1) {
+        *output = static_cast<float>(value);
+      } else {
+        *output = static_cast<float>(value) / static_cast<float>(m_divisor);
+      }
+      return RESULT_OK;
+    }
+    signedValue = static_cast<int>(value);  // negative signed value
+  } else if (negative) {  // negative signed value
+    signedValue = static_cast<int>(value) - (1 << m_bitCount);
+  } else {
+    signedValue = static_cast<int>(value);
+  }
+  if (m_divisor < 0) {
+    *output = static_cast<float>(signedValue) * static_cast<float>(-m_divisor);
+  } else if (m_divisor <= 1) {
+    *output = static_cast<float>(signedValue);
+  } else {
+    *output = static_cast<float>(signedValue) / static_cast<float>(m_divisor);
+  }
+  return RESULT_OK;
+}
+
 result_t NumberDataType::readFromRawValue(unsigned int value,
                                           OutputFormat outputFormat, ostream* output) const {
   size_t length = (m_bitCount < 8) ? 1 : (m_bitCount/8);
@@ -932,6 +996,75 @@ result_t NumberDataType::writeRawValue(unsigned int value, size_t offset, size_t
   return RESULT_OK;
 }
 
+result_t NumberDataType::getRawValueFromFloat(float val, unsigned int* output) const {
+  unsigned int value;
+  if (hasFlag(EXP)) {  // IEEE 754 binary32
+    double dvalue = val;
+    if (m_divisor < 0) {
+      dvalue /= -m_divisor;
+    } else if (m_divisor > 1) {
+      dvalue *= m_divisor;
+    }
+    value = floatToUint(static_cast<float>(dvalue));
+    if (value == 0xffffffff) {
+      return RESULT_ERR_INVALID_NUM;
+    }
+  } else {
+    if (m_divisor == 1) {
+      if (hasFlag(SIG)) {
+        long signedValue = static_cast<long>(val); // TODO static_c?
+        if (signedValue < 0 && m_bitCount != 32) {
+          value = (unsigned int)(signedValue + (1 << m_bitCount));
+        } else {
+          value = (unsigned int)signedValue;
+        }
+      } else if (val < 0) {
+        return RESULT_ERR_INVALID_NUM;  // invalid value
+      } else {
+        value = static_cast<unsigned int>(val);
+      }
+    } else {
+      double dvalue = val;
+      if (m_divisor < 0) {
+        dvalue = round(dvalue / -m_divisor);
+      } else {
+        dvalue = round(dvalue * m_divisor);
+      }
+      int length = static_cast<int>(m_bitCount/8);
+      if (hasFlag(SIG)) {
+        if (dvalue < -exp2((8 * static_cast<double>(length)) - 1)
+            || dvalue >= exp2((8 * static_cast<double>(length)) - 1)) {
+          return RESULT_ERR_OUT_OF_RANGE;  // value out of range
+        }
+        if (dvalue < 0 && m_bitCount != 32) {
+          value = static_cast<unsigned int>(dvalue + (1 << m_bitCount));
+        } else {
+          value = static_cast<unsigned int>(dvalue);
+        }
+      } else {
+        if (dvalue < 0.0 || dvalue >= exp2(8 * static_cast<double>(length))) {
+          return RESULT_ERR_OUT_OF_RANGE;  // value out of range
+        }
+        value = (unsigned int)dvalue;
+      }
+    }
+
+    if (hasFlag(SIG)) {  // signed value
+      if ((value & (1 << (m_bitCount - 1))) != 0) {  // negative signed value
+        if (value < m_minValue) {
+          return RESULT_ERR_OUT_OF_RANGE;  // value out of range
+        }
+      } else if (value > m_maxValue) {
+        return RESULT_ERR_OUT_OF_RANGE;  // value out of range
+      }
+    } else if (value < m_minValue || value > m_maxValue) {
+      return RESULT_ERR_OUT_OF_RANGE;  // value out of range
+    }
+  }
+  *output = value;
+  return RESULT_OK;
+}
+
 result_t NumberDataType::writeSymbols(size_t offset, size_t length, istringstream* input,
                                       SymbolString* output, size_t* usedLength) const {
   unsigned int value;

src/lib/ebus/datatype.h

@@ -539,6 +539,22 @@ class NumberDataType : public DataType {
   result_t readSymbols(size_t offset, size_t length, const SymbolString& input,
       const OutputFormat outputFormat, ostream* output) const override;
 
+  /**
+   * Convert the numeric raw value to its float representation (including optional divisor).
+   * @param value the numeric raw value.
+   * @param output the float variable to write the value to.
+   * @return @a RESULT_OK on success, or an error code.
+   */
+  result_t getFloatFromRawValue(unsigned int value, float* output) const;
+
+  /**
+   * Convert the float value to the numeric raw value (including optional divisor).
+   * @param value the float value.
+   * @param output the variable to write the numeric raw value to.
+   * @return @a RESULT_OK on success, or an error code.
+   */
+  result_t getRawValueFromFloat(float value, unsigned int* output) const;
+
   /**
    * Internal method for interpreting a numeric raw value.
    * @param value the numeric raw value.