git.basschouten.com Git - openhab-addons.git/blob

   1 /**
   2  * Copyright (c) 2010-2023 Contributors to the openHAB project
   3  *
   4  * See the NOTICE file(s) distributed with this work for additional
   5  * information.
   6  *
   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
  10  *
  11  * SPDX-License-Identifier: EPL-2.0
  12  */
  13 package org.openhab.binding.mqtt.homeassistant.internal.handler;
  14
  15 import java.net.URI;
  16 import java.util.ArrayList;
  17 import java.util.Comparator;
  18 import java.util.HashMap;
  19 import java.util.HashSet;
  20 import java.util.List;
  21 import java.util.Map;
  22 import java.util.Objects;
  23 import java.util.Optional;
  24 import java.util.Set;
  25 import java.util.concurrent.CompletableFuture;
  26 import java.util.function.Consumer;
  27 import java.util.stream.Collectors;
  28
  29 import org.eclipse.jdt.annotation.NonNullByDefault;
  30 import org.eclipse.jdt.annotation.Nullable;
  31 import org.openhab.binding.mqtt.generic.AbstractMQTTThingHandler;
  32 import org.openhab.binding.mqtt.generic.ChannelState;
  33 import org.openhab.binding.mqtt.generic.MqttChannelTypeProvider;
  34 import org.openhab.binding.mqtt.generic.TransformationServiceProvider;
  35 import org.openhab.binding.mqtt.generic.tools.DelayedBatchProcessing;
  36 import org.openhab.binding.mqtt.generic.utils.FutureCollector;
  37 import org.openhab.binding.mqtt.homeassistant.generic.internal.MqttBindingConstants;
  38 import org.openhab.binding.mqtt.homeassistant.internal.ComponentChannel;
  39 import org.openhab.binding.mqtt.homeassistant.internal.DiscoverComponents;
  40 import org.openhab.binding.mqtt.homeassistant.internal.DiscoverComponents.ComponentDiscovered;
  41 import org.openhab.binding.mqtt.homeassistant.internal.HaID;
  42 import org.openhab.binding.mqtt.homeassistant.internal.HandlerConfiguration;
  43 import org.openhab.binding.mqtt.homeassistant.internal.component.AbstractComponent;
  44 import org.openhab.binding.mqtt.homeassistant.internal.component.ComponentFactory;
  45 import org.openhab.binding.mqtt.homeassistant.internal.component.Update;
  46 import org.openhab.binding.mqtt.homeassistant.internal.config.ChannelConfigurationTypeAdapterFactory;
  47 import org.openhab.binding.mqtt.homeassistant.internal.exception.ConfigurationException;
  48 import org.openhab.core.config.core.validation.ConfigValidationException;
  49 import org.openhab.core.io.transport.mqtt.MqttBrokerConnection;
  50 import org.openhab.core.thing.Channel;
  51 import org.openhab.core.thing.ChannelGroupUID;
  52 import org.openhab.core.thing.ChannelUID;
  53 import org.openhab.core.thing.Thing;
  54 import org.openhab.core.thing.ThingStatus;
  55 import org.openhab.core.thing.ThingStatusDetail;
  56 import org.openhab.core.thing.ThingTypeUID;
  57 import org.openhab.core.thing.ThingUID;
  58 import org.openhab.core.thing.binding.builder.ThingBuilder;
  59 import org.openhab.core.thing.type.ChannelDefinition;
  60 import org.openhab.core.thing.type.ChannelGroupDefinition;
  61 import org.openhab.core.thing.type.ThingType;
  62 import org.openhab.core.thing.util.ThingHelper;
  63 import org.slf4j.Logger;
  64 import org.slf4j.LoggerFactory;
  65
  66 import com.google.gson.Gson;
  67 import com.google.gson.GsonBuilder;
  68
  69 /**
  70  * Handles HomeAssistant MQTT object things. Such an HA Object can have multiple HA Components with different instances
  71  * of those Components. This handler auto-discovers all available Components and Component Instances and
  72  * adds any new appearing components over time.<br>
  73  * <br>
  74  *
  75  * The specification does not cover the case of disappearing Components. This handler doesn't as well therefore.<br>
  76  * <br>
  77  *
  78  * A Component Instance equals a Channel Group and the Component parts equal Channels.<br>
  79  * <br>
  80  *
  81  * If a Components configuration changes, the known ChannelGroupType and ChannelTypes are replaced with the new ones.
  82  *
  83  * @author David Graeff - Initial contribution
  84  */
  85 @NonNullByDefault
  86 public class HomeAssistantThingHandler extends AbstractMQTTThingHandler
  87         implements ComponentDiscovered, Consumer<List<AbstractComponent<?>>> {
  88     public static final String AVAILABILITY_CHANNEL = "availability";
  89     private static final Comparator<Channel> CHANNEL_COMPARATOR_BY_UID = Comparator
  90             .comparing(channel -> channel.getUID().toString());
  91     private static final URI UPDATABLE_CONFIG_DESCRIPTION_URI = URI.create("thing-type:mqtt:homeassistant-updatable");
  92
  93     private final Logger logger = LoggerFactory.getLogger(HomeAssistantThingHandler.class);
  94
  95     protected final MqttChannelTypeProvider channelTypeProvider;
  96     public final int attributeReceiveTimeout;
  97     protected final DelayedBatchProcessing<AbstractComponent<?>> delayedProcessing;
  98     protected final DiscoverComponents discoverComponents;
  99
 100     private final Gson gson;
 101     protected final Map<@Nullable String, AbstractComponent<?>> haComponents = new HashMap<>();
 102
 103     protected HandlerConfiguration config = new HandlerConfiguration();
 104     private Set<HaID> discoveryHomeAssistantIDs = new HashSet<>();
 105
 106     protected final TransformationServiceProvider transformationServiceProvider;
 107
 108     private boolean started;
 109     private @Nullable Update updateComponent;
 110
 111     /**
 112      * Create a new thing handler for HomeAssistant MQTT components.
 113      * A channel type provider and a topic value receive timeout must be provided.
 114      *
 115      * @param thing The thing of this handler
 116      * @param channelTypeProvider A channel type provider
 117      * @param subscribeTimeout Timeout for the entire tree parsing and subscription. In milliseconds.
 118      * @param attributeReceiveTimeout The timeout per attribute field subscription. In milliseconds.
 119      */
 120     public HomeAssistantThingHandler(Thing thing, MqttChannelTypeProvider channelTypeProvider,
 121             TransformationServiceProvider transformationServiceProvider, int subscribeTimeout,
 122             int attributeReceiveTimeout) {
 123         super(thing, subscribeTimeout);
 124         this.gson = new GsonBuilder().registerTypeAdapterFactory(new ChannelConfigurationTypeAdapterFactory()).create();
 125         this.channelTypeProvider = channelTypeProvider;
 126         this.transformationServiceProvider = transformationServiceProvider;
 127         this.attributeReceiveTimeout = attributeReceiveTimeout;
 128         this.delayedProcessing = new DelayedBatchProcessing<>(attributeReceiveTimeout, this, scheduler);
 129         this.discoverComponents = new DiscoverComponents(thing.getUID(), scheduler, this, this, gson,
 130                 this.transformationServiceProvider);
 131     }
 132
 133     @Override
 134     public void initialize() {
 135         started = false;
 136
 137         config = getConfigAs(HandlerConfiguration.class);
 138         if (config.topics.isEmpty()) {
 139             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_ERROR, "Device topics unknown");
 140             return;
 141         }
 142         discoveryHomeAssistantIDs.addAll(HaID.fromConfig(config));
 143
 144         for (Channel channel : thing.getChannels()) {
 145             final String groupID = channel.getUID().getGroupId();
 146             // Already restored component?
 147             @Nullable
 148             AbstractComponent<?> component = haComponents.get(groupID);
 149             if (component != null) {
 150                 // the types may have been removed in dispose() so we need to add them again
 151                 component.addChannelTypes(channelTypeProvider);
 152                 continue;
 153             }
 154
 155             HaID haID = HaID.fromConfig(config.basetopic, channel.getConfiguration());
 156             discoveryHomeAssistantIDs.add(haID);
 157             ThingUID thingUID = channel.getUID().getThingUID();
 158             String channelConfigurationJSON = (String) channel.getConfiguration().get("config");
 159             if (channelConfigurationJSON == null) {
 160                 logger.warn("Provided channel does not have a 'config' configuration key!");
 161             } else {
 162                 try {
 163                     component = ComponentFactory.createComponent(thingUID, haID, channelConfigurationJSON, this, this,
 164                             scheduler, gson, transformationServiceProvider);
 165                     final ChannelGroupUID groupUID = component.getGroupUID();
 166                     String id = null;
 167                     if (groupUID != null) {
 168                         id = groupUID.getId();
 169                     }
 170                     haComponents.put(id, component);
 171                     component.addChannelTypes(channelTypeProvider);
 172                 } catch (ConfigurationException e) {
 173                     logger.error("Cannot not restore component {}: {}", thing, e.getMessage());
 174                 }
 175             }
 176         }
 177         updateThingType();
 178
 179         super.initialize();
 180     }
 181
 182     @Override
 183     public void dispose() {
 184         // super.dispose() calls stop()
 185         super.dispose();
 186         haComponents.values().forEach(c -> c.removeChannelTypes(channelTypeProvider));
 187     }
 188
 189     @Override
 190     public CompletableFuture<Void> unsubscribeAll() {
 191         // already unsubscribed everything by calling stop()
 192         return CompletableFuture.allOf();
 193     }
 194
 195     /**
 196      * Start a background discovery for the configured HA MQTT object-id.
 197      */
 198     @Override
 199     protected CompletableFuture<@Nullable Void> start(MqttBrokerConnection connection) {
 200         started = true;
 201
 202         connection.setQos(1);
 203         updateStatus(ThingStatus.UNKNOWN);
 204
 205         // Start all known components and channels within the components and put the Thing offline
 206         // if any subscribing failed ( == broker connection lost)
 207         CompletableFuture<@Nullable Void> future = CompletableFuture.allOf(super.start(connection),
 208                 haComponents.values().stream().map(e -> e.start(connection, scheduler, attributeReceiveTimeout))
 209                         .reduce(CompletableFuture.completedFuture(null), (a, v) -> a.thenCompose(b -> v)) // reduce to
 210                                                                                                           // one
 211                         .exceptionally(e -> {
 212                             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_ERROR, e.getMessage());
 213                             return null;
 214                         }));
 215
 216         return future
 217                 .thenCompose(b -> discoverComponents.startDiscovery(connection, 0, discoveryHomeAssistantIDs, this));
 218     }
 219
 220     @Override
 221     protected void stop() {
 222         if (started) {
 223             discoverComponents.stopDiscovery();
 224             delayedProcessing.join();
 225             // haComponents does not need to be synchronised -> the discovery thread is disabled
 226             haComponents.values().stream().map(AbstractComponent::stop) //
 227                     // we need to join all the stops, otherwise they might not be done when start is called
 228                     .collect(FutureCollector.allOf()).join();
 229
 230             started = false;
 231         }
 232         super.stop();
 233     }
 234
 235     @Override
 236     public @Nullable ChannelState getChannelState(ChannelUID channelUID) {
 237         String groupID = channelUID.getGroupId();
 238         AbstractComponent<?> component;
 239         synchronized (haComponents) { // sync whenever discoverComponents is started
 240             component = haComponents.get(groupID);
 241         }
 242         if (component == null) {
 243             return null;
 244         }
 245         ComponentChannel componentChannel = component.getChannel(channelUID.getIdWithoutGroup());
 246         if (componentChannel == null) {
 247             return null;
 248         }
 249         return componentChannel.getState();
 250     }
 251
 252     /**
 253      * Callback of {@link DiscoverComponents}. Add to a delayed batch processor.
 254      */
 255     @Override
 256     public void componentDiscovered(HaID homeAssistantTopicID, AbstractComponent<?> component) {
 257         delayedProcessing.accept(component);
 258     }
 259
 260     /**
 261      * Callback of {@link DelayedBatchProcessing}.
 262      * Add all newly discovered components to the Thing and start the components.
 263      */
 264     @Override
 265     public void accept(List<AbstractComponent<?>> discoveredComponentsList) {
 266         MqttBrokerConnection connection = this.connection;
 267         if (connection == null) {
 268             return;
 269         }
 270
 271         synchronized (haComponents) { // sync whenever discoverComponents is started
 272             for (AbstractComponent<?> discovered : discoveredComponentsList) {
 273                 final ChannelGroupUID groupUID = discovered.getGroupUID();
 274                 String id = null;
 275                 if (groupUID != null) {
 276                     id = groupUID.getId();
 277                 }
 278                 AbstractComponent<?> known = haComponents.get(id);
 279                 // Is component already known?
 280                 if (known != null) {
 281                     if (discovered.getConfigHash() != known.getConfigHash()) {
 282                         // Don't wait for the future to complete. We are also not interested in failures.
 283                         // The component will be replaced in a moment.
 284                         known.stop();
 285                     } else {
 286                         known.setConfigSeen();
 287                         continue;
 288                     }
 289                 }
 290
 291                 // Add channel and group types to the types registry
 292                 discovered.addChannelTypes(channelTypeProvider);
 293                 // Add component to the component map
 294                 haComponents.put(id, discovered);
 295                 // Start component / Subscribe to channel topics
 296                 discovered.start(connection, scheduler, 0).exceptionally(e -> {
 297                     logger.warn("Failed to start component {}", discovered.getHaID(), e);
 298                     return null;
 299                 });
 300
 301                 if (discovered instanceof Update) {
 302                     updateComponent = (Update) discovered;
 303                     updateComponent.setReleaseStateUpdateListener(this::releaseStateUpdated);
 304                 }
 305
 306                 List<Channel> discoveredChannels = discovered.getChannelMap().values().stream()
 307                         .map(ComponentChannel::getChannel).collect(Collectors.toList());
 308                 if (known != null) {
 309                     // We had previously known component with different config hash
 310                     // We remove all conflicting old channels, they will be re-added below based on the new discovery
 311                     logger.debug(
 312                             "Received component {} with slightly different config. Making sure we re-create conflicting channels...",
 313                             discovered.getHaID());
 314                     removeJustRediscoveredChannels(discoveredChannels);
 315                 }
 316
 317                 // Add newly discovered channels. We sort the channels
 318                 // for (mostly) consistent jsondb serialization
 319                 discoveredChannels.sort(CHANNEL_COMPARATOR_BY_UID);
 320                 ThingHelper.addChannelsToThing(thing, discoveredChannels);
 321             }
 322             updateThingType();
 323         }
 324     }
 325
 326     private void removeJustRediscoveredChannels(List<Channel> discoveredChannels) {
 327         ArrayList<Channel> mutableChannels = new ArrayList<>(getThing().getChannels());
 328         Set<ChannelUID> newChannelUIDs = discoveredChannels.stream().map(Channel::getUID).collect(Collectors.toSet());
 329         // Take current channels but remove those channels that were just re-discovered
 330         List<Channel> existingChannelsWithNewlyDiscoveredChannelsRemoved = mutableChannels.stream()
 331                 .filter(existingChannel -> !newChannelUIDs.contains(existingChannel.getUID()))
 332                 .collect(Collectors.toList());
 333         if (existingChannelsWithNewlyDiscoveredChannelsRemoved.size() < mutableChannels.size()) {
 334             // We sort the channels for (mostly) consistent jsondb serialization
 335             existingChannelsWithNewlyDiscoveredChannelsRemoved.sort(CHANNEL_COMPARATOR_BY_UID);
 336             updateThingChannels(existingChannelsWithNewlyDiscoveredChannelsRemoved);
 337         }
 338     }
 339
 340     private void updateThingChannels(List<Channel> channelList) {
 341         ThingBuilder thingBuilder = editThing();
 342         thingBuilder.withChannels(channelList);
 343         updateThing(thingBuilder.build());
 344     }
 345
 346     @Override
 347     protected void updateThingStatus(boolean messageReceived, Optional<Boolean> availabilityTopicsSeen) {
 348         if (availabilityTopicsSeen.orElse(messageReceived)) {
 349             updateStatus(ThingStatus.ONLINE, ThingStatusDetail.NONE);
 350         } else {
 351             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.NONE);
 352         }
 353     }
 354
 355     @Override
 356     public void handleConfigurationUpdate(Map<String, Object> configurationParameters)
 357             throws ConfigValidationException {
 358         if (configurationParameters.containsKey("doUpdate")) {
 359             configurationParameters = new HashMap<>(configurationParameters);
 360             Object value = configurationParameters.remove("doUpdate");
 361             if (value instanceof Boolean doUpdate && doUpdate) {
 362                 Update updateComponent = this.updateComponent;
 363                 if (updateComponent == null) {
 364                     logger.warn(
 365                             "Received update command for Home Assistant device {}, but it does not have an update component.",
 366                             getThing().getUID());
 367                 } else {
 368                     updateComponent.doUpdate();
 369                 }
 370             }
 371         }
 372         super.handleConfigurationUpdate(configurationParameters);
 373     }
 374
 375     private void updateThingType() {
 376         // if this is a dynamic type, then we update the type
 377         ThingTypeUID typeID = thing.getThingTypeUID();
 378         if (!MqttBindingConstants.HOMEASSISTANT_MQTT_THING.equals(typeID)) {
 379             List<ChannelGroupDefinition> groupDefs;
 380             List<ChannelDefinition> channelDefs;
 381             synchronized (haComponents) { // sync whenever discoverComponents is started
 382                 groupDefs = haComponents.values().stream().map(AbstractComponent::getGroupDefinition)
 383                         .filter(Objects::nonNull).map(Objects::requireNonNull).collect(Collectors.toList());
 384                 channelDefs = haComponents.values().stream().map(AbstractComponent::getChannels).flatMap(List::stream)
 385                         .collect(Collectors.toList());
 386             }
 387             var builder = channelTypeProvider.derive(typeID, MqttBindingConstants.HOMEASSISTANT_MQTT_THING)
 388                     .withChannelDefinitions(channelDefs).withChannelGroupDefinitions(groupDefs);
 389             Update updateComponent = this.updateComponent;
 390             if (updateComponent != null && updateComponent.isUpdatable()) {
 391                 builder.withConfigDescriptionURI(UPDATABLE_CONFIG_DESCRIPTION_URI);
 392             }
 393             ThingType thingType = builder.build();
 394
 395             channelTypeProvider.setThingType(typeID, thingType);
 396         }
 397     }
 398
 399     private void releaseStateUpdated(Update.ReleaseState state) {
 400         Map<String, String> properties = editProperties();
 401         properties = state.appendToProperties(properties);
 402         updateProperties(properties);
 403     }
 404 }