2 * Copyright (c) 2010-2023 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);
123 knxChannels.put(channel.getUID(), knxChannel);
124 groupAddresses.addAll(knxChannel.getAllGroupAddresses());
126 if (knxChannel.getChannelType().startsWith("number")) {
127 // check if we need to update the accepted item-type
128 List<InboundSpec> inboundSpecs = knxChannel.getAllGroupAddresses().stream()
129 .map(knxChannel::getListenSpec).filter(Objects::nonNull).map(Objects::requireNonNull).toList();
131 String dpt = inboundSpecs.get(0).getDPT(); // there can be only one DPT on number channels
132 Unit<?> unit = UnitUtils.parseUnit(DPTUnits.getUnitForDpt(dpt));
133 String dimension = unit == null ? null : UnitUtils.getDimensionName(unit);
134 String expectedItemType = dimension == null ? "Number" : "Number:" + dimension; // unknown dimension ->
136 String actualItemType = channel.getAcceptedItemType();
137 if (!expectedItemType.equals(actualItemType)) {
138 ChannelBuilder channelBuilder = callback
139 .createChannelBuilder(channel.getUID(), Objects.requireNonNull(channel.getChannelTypeUID()))
140 .withAcceptedItemType(expectedItemType).withConfiguration(channel.getConfiguration());
141 if (channel.getLabel() != null) {
142 channelBuilder.withLabel(Objects.requireNonNull(channel.getLabel()));
144 if (channel.getDescription() != null) {
145 channelBuilder.withDescription(Objects.requireNonNull(channel.getDescription()));
147 thingBuilder.withoutChannel(channel.getUID());
148 thingBuilder.withChannel(channelBuilder.build());
155 updateThing(thingBuilder.build());
162 public void dispose() {
163 for (ChannelUID channelUID : channelFutures.keySet()) {
164 channelFutures.computeIfPresent(channelUID, (k, v) -> {
170 groupAddresses.clear();
171 groupAddressesWriteBlocked.clear();
172 groupAddressesRespondingSpec.clear();
178 protected void cancelReadFutures() {
179 for (GroupAddress groupAddress : readFutures.keySet()) {
180 readFutures.computeIfPresent(groupAddress, (k, v) -> {
188 public void channelLinked(ChannelUID channelUID) {
189 KNXChannel knxChannel = knxChannels.get(channelUID);
190 if (knxChannel == null) {
191 logger.warn("Channel '{}' received a channel linked event, but no KNXChannel found", channelUID);
194 if (!knxChannel.isControl()) {
195 scheduleRead(knxChannel);
199 protected void scheduleReadJobs() {
201 for (KNXChannel knxChannel : knxChannels.values()) {
202 if (isLinked(knxChannel.getChannelUID()) && !knxChannel.isControl()) {
203 scheduleRead(knxChannel);
208 private void scheduleRead(KNXChannel knxChannel) {
209 List<InboundSpec> readSpecs = knxChannel.getReadSpec();
210 for (InboundSpec readSpec : readSpecs) {
211 readSpec.getGroupAddresses().forEach(ga -> scheduleReadJob(ga, readSpec.getDPT()));
215 private void scheduleReadJob(GroupAddress groupAddress, String dpt) {
216 if (readInterval > 0) {
217 ScheduledFuture<?> future = readFutures.get(groupAddress);
218 if (future == null || future.isDone() || future.isCancelled()) {
219 future = getScheduler().scheduleWithFixedDelay(() -> readDatapoint(groupAddress, dpt), 0, readInterval,
221 readFutures.put(groupAddress, future);
224 getScheduler().submit(() -> readDatapoint(groupAddress, dpt));
228 private void readDatapoint(GroupAddress groupAddress, String dpt) {
229 if (getClient().isConnected()) {
230 if (DPTUtil.getAllowedTypes(dpt).isEmpty()) {
231 logger.warn("DPT '{}' is not supported by the KNX binding", dpt);
234 Datapoint datapoint = new CommandDP(groupAddress, getThing().getUID().toString(), 0, dpt);
235 getClient().readDatapoint(datapoint);
240 public boolean listensTo(GroupAddress destination) {
241 return groupAddresses.contains(destination);
244 /** Handling commands triggered from openHAB */
246 public void handleCommand(ChannelUID channelUID, Command command) {
247 logger.trace("Handling command '{}' for channel '{}'", command, channelUID);
248 KNXChannel knxChannel = knxChannels.get(channelUID);
249 if (knxChannel == null) {
250 logger.warn("Channel '{}' received command, but no KNXChannel found", channelUID);
253 if (command instanceof RefreshType && !knxChannel.isControl()) {
254 logger.debug("Refreshing channel '{}'", channelUID);
255 scheduleRead(knxChannel);
257 if (CHANNEL_RESET.equals(channelUID.getId())) {
258 if (address != null) {
263 OutboundSpec commandSpec = knxChannel.getCommandSpec(command);
264 // only send GroupValueWrite to KNX if GA is not blocked once
265 if (commandSpec != null) {
266 GroupAddress destination = commandSpec.getGroupAddress();
267 if (knxChannel.isControl()) {
268 // always remember, otherwise we might send an old state
269 groupAddressesRespondingSpec.put(destination, commandSpec);
271 if (groupAddressesWriteBlocked.get(destination) != null) {
272 logger.debug("Write to {} blocked for 1s/one call after read.", destination);
273 groupAddressesWriteBlocked.invalidate(destination);
275 getClient().writeToKNX(commandSpec);
279 "None of the configured GAs on channel '{}' could handle the command '{}' of type '{}'",
280 channelUID, command, command.getClass().getSimpleName());
282 } catch (KNXException e) {
283 logger.warn("An error occurred while handling command '{}' on channel '{}': {}", command,
284 channelUID, e.getMessage());
291 private void sendGroupValueResponse(ChannelUID channelUID, GroupAddress destination) {
292 KNXChannel knxChannel = knxChannels.get(channelUID);
293 if (knxChannel == null) {
296 Set<GroupAddress> rsa = knxChannel.getWriteAddresses();
297 if (!rsa.isEmpty()) {
298 logger.trace("onGroupRead size '{}'", rsa.size());
299 OutboundSpec os = groupAddressesRespondingSpec.get(destination);
301 logger.trace("onGroupRead respondToKNX '{}'",
302 os.getGroupAddress()); /* KNXIO: sending real "GroupValueResponse" to the KNX bus. */
304 getClient().respondToKNX(os);
305 } catch (KNXException e) {
306 logger.warn("An error occurred on channel {}: {}", channelUID, e.getMessage(), e);
313 * KNXIO, extended with the ability to respond on "GroupValueRead" telegrams with "GroupValueResponse" telegram
316 public void onGroupRead(AbstractKNXClient client, IndividualAddress source, GroupAddress destination, byte[] asdu) {
317 logger.trace("onGroupRead Thing '{}' received a GroupValueRead telegram from '{}' for destination '{}'",
318 getThing().getUID(), source, destination);
319 for (KNXChannel knxChannel : knxChannels.values()) {
320 if (knxChannel.isControl()) {
321 OutboundSpec responseSpec = knxChannel.getResponseSpec(destination, RefreshType.REFRESH);
322 if (responseSpec != null) {
323 logger.trace("onGroupRead isControl -> postCommand");
324 // This event should be sent to KNX as GroupValueResponse immediately.
325 sendGroupValueResponse(knxChannel.getChannelUID(), destination);
327 // block write attempts for 1s or 1 request to prevent loops
328 if (!groupAddressesWriteBlocked.containsKey(destination)) {
329 groupAddressesWriteBlocked.put(destination, () -> null);
331 groupAddressesWriteBlocked.putValue(destination, true);
333 // Send REFRESH to openHAB to get this event for scripting with postCommand
334 // and remember to ignore/block this REFRESH to be sent back to KNX as GroupValueWrite after
335 // postCommand is done!
336 postCommand(knxChannel.getChannelUID(), RefreshType.REFRESH);
343 public void onGroupReadResponse(AbstractKNXClient client, IndividualAddress source, GroupAddress destination,
345 // GroupValueResponses are treated the same as GroupValueWrite telegrams
346 logger.trace("onGroupReadResponse Thing '{}' processes a GroupValueResponse telegram for destination '{}'",
347 getThing().getUID(), destination);
348 onGroupWrite(client, source, destination, asdu);
352 * KNXIO, here value changes are set, coming from KNX OR openHAB.
355 public void onGroupWrite(AbstractKNXClient client, IndividualAddress source, GroupAddress destination,
357 logger.debug("onGroupWrite Thing '{}' received a GroupValueWrite telegram from '{}' for destination '{}'",
358 getThing().getUID(), source, destination);
360 for (KNXChannel knxChannel : knxChannels.values()) {
361 InboundSpec listenSpec = knxChannel.getListenSpec(destination);
362 if (listenSpec != null) {
364 "onGroupWrite Thing '{}' processes a GroupValueWrite telegram for destination '{}' for channel '{}'",
365 getThing().getUID(), destination, knxChannel.getChannelUID());
367 * Remember current KNXIO outboundSpec only if it is a control channel.
369 if (knxChannel.isControl()) {
370 logger.trace("onGroupWrite isControl");
371 Type value = ValueDecoder.decode(listenSpec.getDPT(), asdu, knxChannel.preferredType());
373 OutboundSpec commandSpec = knxChannel.getCommandSpec(value);
374 if (commandSpec != null) {
375 groupAddressesRespondingSpec.put(destination, commandSpec);
379 processDataReceived(destination, asdu, listenSpec, knxChannel);
384 private void processDataReceived(GroupAddress destination, byte[] asdu, InboundSpec listenSpec,
385 KNXChannel knxChannel) {
386 if (DPTUtil.getAllowedTypes(listenSpec.getDPT()).isEmpty()) {
387 logger.warn("DPT '{}' is not supported by the KNX binding.", listenSpec.getDPT());
391 Type value = ValueDecoder.decode(listenSpec.getDPT(), asdu, knxChannel.preferredType());
393 if (knxChannel.isControl()) {
394 ChannelUID channelUID = knxChannel.getChannelUID();
396 if (KNXBindingConstants.CHANNEL_DIMMER_CONTROL.equals(knxChannel.getChannelType())) {
397 // if we have a dimmer control channel, check if a frequency is defined
398 Channel channel = getThing().getChannel(channelUID);
399 if (channel == null) {
400 logger.warn("Failed to find channel for ChannelUID '{}'", channelUID);
403 frequency = ((BigDecimal) Objects.requireNonNullElse(
404 channel.getConfiguration().get(KNXBindingConstants.REPEAT_FREQUENCY), BigDecimal.ZERO))
407 // disable dimming by binding
410 if ((value instanceof UnDefType || value instanceof IncreaseDecreaseType) && frequency > 0) {
411 // continuous dimming by the binding
412 // cancel a running scheduler before adding a new (and only add if not UnDefType)
413 ScheduledFuture<?> oldFuture = channelFutures.remove(channelUID);
414 if (oldFuture != null) {
415 oldFuture.cancel(true);
417 if (value instanceof IncreaseDecreaseType type) {
418 channelFutures.put(channelUID, scheduler.scheduleWithFixedDelay(
419 () -> postCommand(channelUID, type), 0, frequency, TimeUnit.MILLISECONDS));
422 if (value instanceof Command command) {
423 logger.trace("processDataReceived postCommand new value '{}' for GA '{}'", asdu, address);
424 postCommand(channelUID, command);
428 if (value instanceof State state && !(value instanceof UnDefType)) {
429 updateState(knxChannel.getChannelUID(), state);
434 "Ignoring KNX bus data for channel '{}': couldn't transform to any Type (GA='{}', DPT='{}', data='{}')",
435 knxChannel.getChannelUID(), destination, listenSpec.getDPT(), HexUtils.bytesToHex(asdu));
439 protected final ScheduledExecutorService getScheduler() {
440 return getBridgeHandler().getScheduler();
443 protected final ScheduledExecutorService getBackgroundScheduler() {
444 return getBridgeHandler().getBackgroundScheduler();
447 protected final KNXBridgeBaseThingHandler getBridgeHandler() {
448 Bridge bridge = getBridge();
449 if (bridge != null) {
450 KNXBridgeBaseThingHandler handler = (KNXBridgeBaseThingHandler) bridge.getHandler();
451 if (handler != null) {
455 throw new IllegalStateException("The bridge must not be null and must be initialized");
458 protected final KNXClient getClient() {
459 return getBridgeHandler().getClient();
462 protected final boolean describeDevice(@Nullable IndividualAddress address) {
463 if (address == null) {
466 DeviceInspector inspector = new DeviceInspector(getClient().getDeviceInfoClient(), address);
467 DeviceInspector.Result result = inspector.readDeviceInfo();
468 if (result != null) {
469 Map<String, String> properties = editProperties();
470 properties.putAll(result.getProperties());
471 updateProperties(properties);
477 protected final void restart() {
478 if (address != null) {
479 getClient().restartNetworkDevice(address);
484 public void bridgeStatusChanged(ThingStatusInfo bridgeStatusInfo) {
485 if (bridgeStatusInfo.getStatus() == ThingStatus.ONLINE) {
487 } else if (bridgeStatusInfo.getStatus() == ThingStatus.OFFLINE) {
489 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.BRIDGE_OFFLINE);
493 private void pollDeviceStatus() {
495 if (address != null && getClient().isConnected()) {
496 logger.debug("Polling individual address '{}'", address);
497 boolean isReachable = getClient().isReachable(address);
499 updateStatus(ThingStatus.ONLINE);
500 DeviceConfig config = getConfigAs(DeviceConfig.class);
501 if (!filledDescription && config.getFetch()) {
502 Future<?> descriptionJob = this.descriptionJob;
503 if (descriptionJob == null || descriptionJob.isCancelled()) {
504 long initialDelay = Math.round(config.getPingInterval() * random.nextFloat());
505 this.descriptionJob = getBackgroundScheduler().schedule(() -> {
506 filledDescription = describeDevice(address);
507 }, initialDelay, TimeUnit.SECONDS);
511 updateStatus(ThingStatus.OFFLINE);
514 } catch (KNXException e) {
515 logger.debug("An error occurred while testing the reachability of a thing '{}': {}", getThing().getUID(),
517 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR,
518 KNXTranslationProvider.I18N.getLocalizedException(e));
522 protected void attachToClient() {
523 if (!getClient().isConnected()) {
524 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.BRIDGE_OFFLINE);
527 DeviceConfig config = getConfigAs(DeviceConfig.class);
529 if (!config.getAddress().isEmpty()) {
530 updateStatus(ThingStatus.UNKNOWN);
531 address = new IndividualAddress(config.getAddress());
533 long pingInterval = config.getPingInterval();
534 long initialPingDelay = Math.round(INITIAL_PING_DELAY * random.nextFloat());
536 ScheduledFuture<?> pollingJob = this.pollingJob;
537 if ((pollingJob == null || pollingJob.isCancelled())) {
538 logger.debug("'{}' will be polled every {}s", getThing().getUID(), pingInterval);
539 this.pollingJob = getBackgroundScheduler().scheduleWithFixedDelay(this::pollDeviceStatus,
540 initialPingDelay, pingInterval, TimeUnit.SECONDS);
543 updateStatus(ThingStatus.ONLINE);
545 } catch (KNXFormatException e) {
546 logger.debug("An exception occurred while setting the individual address '{}': {}", config.getAddress(),
548 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_ERROR,
549 KNXTranslationProvider.I18N.getLocalizedException(e));
551 getClient().registerGroupAddressListener(this);
555 protected void detachFromClient() {
556 ScheduledFuture<?> pollingJobSynced = pollingJob;
557 if (pollingJobSynced != null) {
558 pollingJobSynced.cancel(true);
561 ScheduledFuture<?> descriptionJobSynced = descriptionJob;
562 if (descriptionJobSynced != null) {
563 descriptionJobSynced.cancel(true);
564 descriptionJob = null;
567 Bridge bridge = getBridge();
568 if (bridge != null) {
569 KNXBridgeBaseThingHandler handler = (KNXBridgeBaseThingHandler) bridge.getHandler();
570 if (handler != null) {
571 handler.getClient().unregisterGroupAddressListener(this);