2 * Copyright (c) 2010-2024 Contributors to the openHAB project
4 * See the NOTICE file(s) distributed with this work for additional
7 * This program and the accompanying materials are made available under the
8 * terms of the Eclipse Public License 2.0 which is available at
9 * http://www.eclipse.org/legal/epl-2.0
11 * SPDX-License-Identifier: EPL-2.0
13 package org.openhab.binding.knx.internal.handler;
15 import static org.openhab.binding.knx.internal.KNXBindingConstants.*;
17 import java.math.BigDecimal;
18 import java.time.Duration;
19 import java.util.List;
21 import java.util.Objects;
22 import java.util.Random;
24 import java.util.concurrent.ConcurrentHashMap;
25 import java.util.concurrent.Future;
26 import java.util.concurrent.ScheduledExecutorService;
27 import java.util.concurrent.ScheduledFuture;
28 import java.util.concurrent.TimeUnit;
30 import javax.measure.Unit;
32 import org.eclipse.jdt.annotation.NonNullByDefault;
33 import org.eclipse.jdt.annotation.Nullable;
34 import org.openhab.binding.knx.internal.KNXBindingConstants;
35 import org.openhab.binding.knx.internal.channel.KNXChannel;
36 import org.openhab.binding.knx.internal.channel.KNXChannelFactory;
37 import org.openhab.binding.knx.internal.client.AbstractKNXClient;
38 import org.openhab.binding.knx.internal.client.DeviceInspector;
39 import org.openhab.binding.knx.internal.client.InboundSpec;
40 import org.openhab.binding.knx.internal.client.KNXClient;
41 import org.openhab.binding.knx.internal.client.OutboundSpec;
42 import org.openhab.binding.knx.internal.config.DeviceConfig;
43 import org.openhab.binding.knx.internal.dpt.DPTUnits;
44 import org.openhab.binding.knx.internal.dpt.DPTUtil;
45 import org.openhab.binding.knx.internal.dpt.ValueDecoder;
46 import org.openhab.binding.knx.internal.i18n.KNXTranslationProvider;
47 import org.openhab.core.cache.ExpiringCacheMap;
48 import org.openhab.core.library.types.IncreaseDecreaseType;
49 import org.openhab.core.thing.Bridge;
50 import org.openhab.core.thing.Channel;
51 import org.openhab.core.thing.ChannelUID;
52 import org.openhab.core.thing.Thing;
53 import org.openhab.core.thing.ThingStatus;
54 import org.openhab.core.thing.ThingStatusDetail;
55 import org.openhab.core.thing.ThingStatusInfo;
56 import org.openhab.core.thing.binding.BaseThingHandler;
57 import org.openhab.core.thing.binding.ThingHandlerCallback;
58 import org.openhab.core.thing.binding.builder.ChannelBuilder;
59 import org.openhab.core.thing.binding.builder.ThingBuilder;
60 import org.openhab.core.types.Command;
61 import org.openhab.core.types.RefreshType;
62 import org.openhab.core.types.State;
63 import org.openhab.core.types.Type;
64 import org.openhab.core.types.UnDefType;
65 import org.openhab.core.types.util.UnitUtils;
66 import org.openhab.core.util.HexUtils;
67 import org.slf4j.Logger;
68 import org.slf4j.LoggerFactory;
70 import tuwien.auto.calimero.GroupAddress;
71 import tuwien.auto.calimero.IndividualAddress;
72 import tuwien.auto.calimero.KNXException;
73 import tuwien.auto.calimero.KNXFormatException;
74 import tuwien.auto.calimero.datapoint.CommandDP;
75 import tuwien.auto.calimero.datapoint.Datapoint;
78 * The {@link DeviceThingHandler} is responsible for handling commands and state updates sent to and received from the
79 * bus and updating the channels correspondingly.
81 * @author Simon Kaufmann - Initial contribution and API
82 * @author Jan N. Klug - Refactored for performance
85 public class DeviceThingHandler extends BaseThingHandler implements GroupAddressListener {
86 private static final int INITIAL_PING_DELAY = 5;
87 private final Logger logger = LoggerFactory.getLogger(DeviceThingHandler.class);
89 private final Set<GroupAddress> groupAddresses = ConcurrentHashMap.newKeySet();
90 private final ExpiringCacheMap<GroupAddress, @Nullable Boolean> groupAddressesWriteBlocked = new ExpiringCacheMap<>(
91 Duration.ofMillis(1000));
92 private final Map<GroupAddress, OutboundSpec> groupAddressesRespondingSpec = new ConcurrentHashMap<>();
93 private final Map<GroupAddress, ScheduledFuture<?>> readFutures = new ConcurrentHashMap<>();
94 private final Map<ChannelUID, ScheduledFuture<?>> channelFutures = new ConcurrentHashMap<>();
95 private final Map<ChannelUID, KNXChannel> knxChannels = new ConcurrentHashMap<>();
96 private final Random random = new Random();
97 protected @Nullable IndividualAddress address;
98 private int readInterval;
99 private @Nullable ScheduledFuture<?> descriptionJob;
100 private boolean filledDescription = false;
101 private @Nullable ScheduledFuture<?> pollingJob;
103 public DeviceThingHandler(Thing thing) {
108 public void initialize() {
109 DeviceConfig config = getConfigAs(DeviceConfig.class);
110 readInterval = config.getReadInterval();
112 // gather all GAs from channel configurations and create channels
113 ThingBuilder thingBuilder = editThing();
114 boolean modified = false;
115 ThingHandlerCallback callback = getCallback();
116 if (callback == null) {
117 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.NONE, "Framework failure: callback must not be null");
121 for (Channel channel : getThing().getChannels()) {
122 KNXChannel knxChannel = KNXChannelFactory.createKnxChannel(channel);
124 if (knxChannel.getChannelType().startsWith("number")) {
125 // check if we need to update the accepted item-type
126 List<InboundSpec> inboundSpecs = knxChannel.getAllGroupAddresses().stream()
127 .map(knxChannel::getListenSpec).filter(Objects::nonNull).map(Objects::requireNonNull).toList();
128 if (inboundSpecs.isEmpty()) {
129 logger.warn("Skipping {}: group address / DPT not according to Group Address Notation",
134 String dpt = inboundSpecs.get(0).getDPT(); // there can be only one DPT on number channels
135 Unit<?> unit = UnitUtils.parseUnit(DPTUnits.getUnitForDpt(dpt));
136 String dimension = unit == null ? null : UnitUtils.getDimensionName(unit);
137 String expectedItemType = dimension == null ? "Number" : "Number:" + dimension; // unknown dimension ->
139 String actualItemType = channel.getAcceptedItemType();
140 if (!expectedItemType.equals(actualItemType)) {
141 ChannelBuilder channelBuilder = callback
142 .createChannelBuilder(channel.getUID(), Objects.requireNonNull(channel.getChannelTypeUID()))
143 .withAcceptedItemType(expectedItemType).withConfiguration(channel.getConfiguration());
144 if (channel.getLabel() != null) {
145 channelBuilder.withLabel(Objects.requireNonNull(channel.getLabel()));
147 if (channel.getDescription() != null) {
148 channelBuilder.withDescription(Objects.requireNonNull(channel.getDescription()));
150 thingBuilder.withoutChannel(channel.getUID());
151 thingBuilder.withChannel(channelBuilder.build());
156 // add channels only if they could be successfully processed
157 knxChannels.put(channel.getUID(), knxChannel);
158 groupAddresses.addAll(knxChannel.getAllGroupAddresses());
162 updateThing(thingBuilder.build());
169 public void dispose() {
170 for (ChannelUID channelUID : channelFutures.keySet()) {
171 channelFutures.computeIfPresent(channelUID, (k, v) -> {
177 groupAddresses.clear();
178 groupAddressesWriteBlocked.clear();
179 groupAddressesRespondingSpec.clear();
185 protected void cancelReadFutures() {
186 for (GroupAddress groupAddress : readFutures.keySet()) {
187 readFutures.computeIfPresent(groupAddress, (k, v) -> {
195 public void channelLinked(ChannelUID channelUID) {
196 KNXChannel knxChannel = knxChannels.get(channelUID);
197 if (knxChannel == null) {
198 logger.warn("Channel '{}' received a channel linked event, but no KNXChannel found", channelUID);
201 if (!knxChannel.isControl()) {
202 scheduleRead(knxChannel);
206 protected void scheduleReadJobs() {
208 for (KNXChannel knxChannel : knxChannels.values()) {
209 if (isLinked(knxChannel.getChannelUID()) && !knxChannel.isControl()) {
210 scheduleRead(knxChannel);
215 private void scheduleRead(KNXChannel knxChannel) {
216 List<InboundSpec> readSpecs = knxChannel.getReadSpec();
217 for (InboundSpec readSpec : readSpecs) {
218 readSpec.getGroupAddresses().forEach(ga -> scheduleReadJob(ga, readSpec.getDPT()));
222 private void scheduleReadJob(GroupAddress groupAddress, String dpt) {
223 if (readInterval > 0) {
224 ScheduledFuture<?> future = readFutures.get(groupAddress);
225 if (future == null || future.isDone() || future.isCancelled()) {
226 future = getScheduler().scheduleWithFixedDelay(() -> readDatapoint(groupAddress, dpt), 0, readInterval,
228 readFutures.put(groupAddress, future);
231 getScheduler().submit(() -> readDatapoint(groupAddress, dpt));
235 private void readDatapoint(GroupAddress groupAddress, String dpt) {
236 if (getClient().isConnected()) {
237 if (DPTUtil.getAllowedTypes(dpt).isEmpty()) {
238 logger.warn("DPT '{}' is not supported by the KNX binding", dpt);
241 Datapoint datapoint = new CommandDP(groupAddress, getThing().getUID().toString(), 0, dpt);
242 getClient().readDatapoint(datapoint);
247 public boolean listensTo(GroupAddress destination) {
248 return groupAddresses.contains(destination);
251 /** Handling commands triggered from openHAB */
253 public void handleCommand(ChannelUID channelUID, Command command) {
254 logger.trace("Handling command '{}' for channel '{}'", command, channelUID);
255 KNXChannel knxChannel = knxChannels.get(channelUID);
256 if (knxChannel == null) {
257 logger.warn("Channel '{}' received command, but no KNXChannel found", channelUID);
260 if (command instanceof RefreshType && !knxChannel.isControl()) {
261 logger.debug("Refreshing channel '{}'", channelUID);
262 scheduleRead(knxChannel);
264 if (CHANNEL_RESET.equals(channelUID.getId())) {
265 if (address != null) {
270 OutboundSpec commandSpec = knxChannel.getCommandSpec(command);
271 // only send GroupValueWrite to KNX if GA is not blocked once
272 if (commandSpec != null) {
273 GroupAddress destination = commandSpec.getGroupAddress();
274 if (knxChannel.isControl()) {
275 // always remember, otherwise we might send an old state
276 groupAddressesRespondingSpec.put(destination, commandSpec);
278 if (groupAddressesWriteBlocked.get(destination) != null) {
279 logger.debug("Write to {} blocked for 1s/one call after read.", destination);
280 groupAddressesWriteBlocked.invalidate(destination);
282 getClient().writeToKNX(commandSpec);
286 "None of the configured GAs on channel '{}' could handle the command '{}' of type '{}'",
287 channelUID, command, command.getClass().getSimpleName());
289 } catch (KNXException e) {
290 logger.warn("An error occurred while handling command '{}' on channel '{}': {}", command,
291 channelUID, e.getMessage());
298 private void sendGroupValueResponse(ChannelUID channelUID, GroupAddress destination) {
299 KNXChannel knxChannel = knxChannels.get(channelUID);
300 if (knxChannel == null) {
303 List<GroupAddress> rsa = knxChannel.getWriteAddresses();
304 if (!rsa.isEmpty()) {
305 logger.trace("onGroupRead size '{}'", rsa.size());
306 OutboundSpec os = groupAddressesRespondingSpec.get(destination);
308 logger.trace("onGroupRead respondToKNX '{}'",
309 os.getGroupAddress()); /* KNXIO: sending real "GroupValueResponse" to the KNX bus. */
311 getClient().respondToKNX(os);
312 } catch (KNXException e) {
313 logger.warn("An error occurred on channel {}: {}", channelUID, e.getMessage(), e);
320 * KNXIO, extended with the ability to respond on "GroupValueRead" telegrams with "GroupValueResponse" telegram
323 public void onGroupRead(AbstractKNXClient client, IndividualAddress source, GroupAddress destination, byte[] asdu) {
324 logger.trace("onGroupRead Thing '{}' received a GroupValueRead telegram from '{}' for destination '{}'",
325 getThing().getUID(), source, destination);
326 for (KNXChannel knxChannel : knxChannels.values()) {
327 if (knxChannel.isControl()) {
328 OutboundSpec responseSpec = knxChannel.getResponseSpec(destination, RefreshType.REFRESH);
329 if (responseSpec != null) {
330 logger.trace("onGroupRead isControl -> postCommand");
331 // This event should be sent to KNX as GroupValueResponse immediately.
332 sendGroupValueResponse(knxChannel.getChannelUID(), destination);
334 // block write attempts for 1s or 1 request to prevent loops
335 if (!groupAddressesWriteBlocked.containsKey(destination)) {
336 groupAddressesWriteBlocked.put(destination, () -> null);
338 groupAddressesWriteBlocked.putValue(destination, true);
340 // Send REFRESH to openHAB to get this event for scripting with postCommand
341 // and remember to ignore/block this REFRESH to be sent back to KNX as GroupValueWrite after
342 // postCommand is done!
343 postCommand(knxChannel.getChannelUID(), RefreshType.REFRESH);
350 public void onGroupReadResponse(AbstractKNXClient client, IndividualAddress source, GroupAddress destination,
352 // GroupValueResponses are treated the same as GroupValueWrite telegrams
353 logger.trace("onGroupReadResponse Thing '{}' processes a GroupValueResponse telegram for destination '{}'",
354 getThing().getUID(), destination);
355 onGroupWrite(client, source, destination, asdu);
359 * KNXIO, here value changes are set, coming from KNX OR openHAB.
362 public void onGroupWrite(AbstractKNXClient client, IndividualAddress source, GroupAddress destination,
364 logger.debug("onGroupWrite Thing '{}' received a GroupValueWrite telegram from '{}' for destination '{}'",
365 getThing().getUID(), source, destination);
367 for (KNXChannel knxChannel : knxChannels.values()) {
368 InboundSpec listenSpec = knxChannel.getListenSpec(destination);
369 if (listenSpec != null) {
371 "onGroupWrite Thing '{}' processes a GroupValueWrite telegram for destination '{}' for channel '{}'",
372 getThing().getUID(), destination, knxChannel.getChannelUID());
374 * Remember current KNXIO outboundSpec only if it is a control channel.
376 if (knxChannel.isControl()) {
377 logger.trace("onGroupWrite isControl");
378 Type value = ValueDecoder.decode(listenSpec.getDPT(), asdu, knxChannel.preferredType());
380 OutboundSpec commandSpec = knxChannel.getCommandSpec(value);
381 if (commandSpec != null) {
382 groupAddressesRespondingSpec.put(destination, commandSpec);
386 processDataReceived(destination, asdu, listenSpec, knxChannel);
391 private void processDataReceived(GroupAddress destination, byte[] asdu, InboundSpec listenSpec,
392 KNXChannel knxChannel) {
393 if (DPTUtil.getAllowedTypes(listenSpec.getDPT()).isEmpty()) {
394 logger.warn("DPT '{}' is not supported by the KNX binding.", listenSpec.getDPT());
398 Type value = ValueDecoder.decode(listenSpec.getDPT(), asdu, knxChannel.preferredType());
400 if (knxChannel.isControl()) {
401 ChannelUID channelUID = knxChannel.getChannelUID();
403 if (KNXBindingConstants.CHANNEL_DIMMER_CONTROL.equals(knxChannel.getChannelType())) {
404 // if we have a dimmer control channel, check if a frequency is defined
405 Channel channel = getThing().getChannel(channelUID);
406 if (channel == null) {
407 logger.warn("Failed to find channel for ChannelUID '{}'", channelUID);
410 frequency = ((BigDecimal) Objects.requireNonNullElse(
411 channel.getConfiguration().get(KNXBindingConstants.REPEAT_FREQUENCY), BigDecimal.ZERO))
414 // disable dimming by binding
417 if ((value instanceof UnDefType || value instanceof IncreaseDecreaseType) && frequency > 0) {
418 // continuous dimming by the binding
419 // cancel a running scheduler before adding a new (and only add if not UnDefType)
420 ScheduledFuture<?> oldFuture = channelFutures.remove(channelUID);
421 if (oldFuture != null) {
422 oldFuture.cancel(true);
424 if (value instanceof IncreaseDecreaseType increaseDecreaseCommand) {
425 channelFutures.put(channelUID,
426 scheduler.scheduleWithFixedDelay(() -> postCommand(channelUID, increaseDecreaseCommand),
427 0, frequency, TimeUnit.MILLISECONDS));
430 if (value instanceof Command command) {
431 logger.trace("processDataReceived postCommand to channel '{}' new value '{}' for GA '{}'",
432 channelUID, asdu, destination);
433 postCommand(channelUID, command);
437 if (value instanceof State state && !(value instanceof UnDefType)) {
438 logger.trace("processDataReceived updateState to channel '{}' new value '{}' for GA '{}'",
439 knxChannel.getChannelUID(), value, destination);
440 updateState(knxChannel.getChannelUID(), state);
445 "Ignoring KNX bus data for channel '{}': couldn't transform to any Type (GA='{}', DPT='{}', data='{}')",
446 knxChannel.getChannelUID(), destination, listenSpec.getDPT(), HexUtils.bytesToHex(asdu));
450 protected final ScheduledExecutorService getScheduler() {
451 return getBridgeHandler().getScheduler();
454 protected final ScheduledExecutorService getBackgroundScheduler() {
455 return getBridgeHandler().getBackgroundScheduler();
458 protected final KNXBridgeBaseThingHandler getBridgeHandler() {
459 Bridge bridge = getBridge();
460 if (bridge != null) {
461 KNXBridgeBaseThingHandler handler = (KNXBridgeBaseThingHandler) bridge.getHandler();
462 if (handler != null) {
466 throw new IllegalStateException("The bridge must not be null and must be initialized");
469 protected final KNXClient getClient() {
470 return getBridgeHandler().getClient();
473 protected final boolean describeDevice(@Nullable IndividualAddress address) {
474 if (address == null) {
477 DeviceInspector inspector = new DeviceInspector(getClient().getDeviceInfoClient(), address);
478 DeviceInspector.Result result = inspector.readDeviceInfo();
479 if (result != null) {
480 Map<String, String> properties = editProperties();
481 properties.putAll(result.getProperties());
482 updateProperties(properties);
488 protected final void restart() {
489 if (address != null) {
490 getClient().restartNetworkDevice(address);
495 public void bridgeStatusChanged(ThingStatusInfo bridgeStatusInfo) {
496 if (bridgeStatusInfo.getStatus() == ThingStatus.ONLINE) {
498 } else if (bridgeStatusInfo.getStatus() == ThingStatus.OFFLINE) {
500 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.BRIDGE_OFFLINE);
504 private void pollDeviceStatus() {
506 if (address != null && getClient().isConnected()) {
507 logger.debug("Polling individual address '{}'", address);
508 boolean isReachable = getClient().isReachable(address);
510 updateStatus(ThingStatus.ONLINE);
511 DeviceConfig config = getConfigAs(DeviceConfig.class);
512 if (!filledDescription && config.getFetch()) {
513 Future<?> descriptionJob = this.descriptionJob;
514 if (descriptionJob == null || descriptionJob.isCancelled()) {
515 long initialDelay = Math.round(config.getPingInterval() * random.nextFloat());
516 this.descriptionJob = getBackgroundScheduler().schedule(() -> {
517 filledDescription = describeDevice(address);
518 }, initialDelay, TimeUnit.SECONDS);
522 updateStatus(ThingStatus.OFFLINE);
525 } catch (KNXException e) {
526 logger.debug("An error occurred while testing the reachability of a thing '{}': {}", getThing().getUID(),
528 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR,
529 KNXTranslationProvider.I18N.getLocalizedException(e));
533 protected void attachToClient() {
534 if (!getClient().isConnected()) {
535 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.BRIDGE_OFFLINE);
538 DeviceConfig config = getConfigAs(DeviceConfig.class);
540 if (!config.getAddress().isEmpty()) {
541 updateStatus(ThingStatus.UNKNOWN);
542 address = new IndividualAddress(config.getAddress());
544 long pingInterval = config.getPingInterval();
545 long initialPingDelay = Math.round(INITIAL_PING_DELAY * random.nextFloat());
547 ScheduledFuture<?> pollingJob = this.pollingJob;
548 if ((pollingJob == null || pollingJob.isCancelled())) {
549 logger.debug("'{}' will be polled every {}s", getThing().getUID(), pingInterval);
550 this.pollingJob = getBackgroundScheduler().scheduleWithFixedDelay(this::pollDeviceStatus,
551 initialPingDelay, pingInterval, TimeUnit.SECONDS);
554 updateStatus(ThingStatus.ONLINE);
556 } catch (KNXFormatException e) {
557 logger.debug("An exception occurred while setting the individual address '{}': {}", config.getAddress(),
559 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_ERROR,
560 KNXTranslationProvider.I18N.getLocalizedException(e));
562 getClient().registerGroupAddressListener(this);
566 protected void detachFromClient() {
567 ScheduledFuture<?> pollingJobSynced = pollingJob;
568 if (pollingJobSynced != null) {
569 pollingJobSynced.cancel(true);
572 ScheduledFuture<?> descriptionJobSynced = descriptionJob;
573 if (descriptionJobSynced != null) {
574 descriptionJobSynced.cancel(true);
575 descriptionJob = null;
578 Bridge bridge = getBridge();
579 if (bridge != null) {
580 KNXBridgeBaseThingHandler handler = (KNXBridgeBaseThingHandler) bridge.getHandler();
581 if (handler != null) {
582 handler.getClient().unregisterGroupAddressListener(this);