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

   1 /**
   2  * Copyright (c) 2010-2024 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.Comparator;
  17 import java.util.HashMap;
  18 import java.util.HashSet;
  19 import java.util.List;
  20 import java.util.Map;
  21 import java.util.Objects;
  22 import java.util.Optional;
  23 import java.util.Set;
  24 import java.util.concurrent.CompletableFuture;
  25 import java.util.function.Consumer;
  26
  27 import org.eclipse.jdt.annotation.NonNullByDefault;
  28 import org.eclipse.jdt.annotation.Nullable;
  29 import org.openhab.binding.mqtt.generic.AbstractMQTTThingHandler;
  30 import org.openhab.binding.mqtt.generic.ChannelState;
  31 import org.openhab.binding.mqtt.generic.MqttChannelStateDescriptionProvider;
  32 import org.openhab.binding.mqtt.generic.MqttChannelTypeProvider;
  33 import org.openhab.binding.mqtt.generic.tools.DelayedBatchProcessing;
  34 import org.openhab.binding.mqtt.generic.utils.FutureCollector;
  35 import org.openhab.binding.mqtt.homeassistant.generic.internal.MqttBindingConstants;
  36 import org.openhab.binding.mqtt.homeassistant.internal.DiscoverComponents;
  37 import org.openhab.binding.mqtt.homeassistant.internal.DiscoverComponents.ComponentDiscovered;
  38 import org.openhab.binding.mqtt.homeassistant.internal.HaID;
  39 import org.openhab.binding.mqtt.homeassistant.internal.HandlerConfiguration;
  40 import org.openhab.binding.mqtt.homeassistant.internal.component.AbstractComponent;
  41 import org.openhab.binding.mqtt.homeassistant.internal.component.ComponentFactory;
  42 import org.openhab.binding.mqtt.homeassistant.internal.component.Update;
  43 import org.openhab.binding.mqtt.homeassistant.internal.config.ChannelConfigurationTypeAdapterFactory;
  44 import org.openhab.binding.mqtt.homeassistant.internal.exception.ConfigurationException;
  45 import org.openhab.core.config.core.Configuration;
  46 import org.openhab.core.config.core.validation.ConfigValidationException;
  47 import org.openhab.core.io.transport.mqtt.MqttBrokerConnection;
  48 import org.openhab.core.thing.Channel;
  49 import org.openhab.core.thing.ChannelUID;
  50 import org.openhab.core.thing.Thing;
  51 import org.openhab.core.thing.ThingStatus;
  52 import org.openhab.core.thing.ThingStatusDetail;
  53 import org.openhab.core.thing.ThingTypeUID;
  54 import org.openhab.core.thing.ThingUID;
  55 import org.openhab.core.thing.binding.builder.ThingBuilder;
  56 import org.openhab.core.thing.type.ChannelTypeRegistry;
  57 import org.slf4j.Logger;
  58 import org.slf4j.LoggerFactory;
  59
  60 import com.google.gson.Gson;
  61 import com.google.gson.GsonBuilder;
  62 import com.hubspot.jinjava.Jinjava;
  63
  64 /**
  65  * Handles HomeAssistant MQTT object things. Such an HA Object can have multiple HA Components with different instances
  66  * of those Components. This handler auto-discovers all available Components and Component Instances and
  67  * adds any new appearing components over time.<br>
  68  * <br>
  69  *
  70  * The specification does not cover the case of disappearing Components. This handler doesn't as well therefore.<br>
  71  * <br>
  72  *
  73  * A Component Instance equals a Channel Group and the Component parts equal Channels.<br>
  74  * <br>
  75  *
  76  * If a Components configuration changes, the known ChannelGroupType and ChannelTypes are replaced with the new ones.
  77  *
  78  * @author David Graeff - Initial contribution
  79  */
  80 @NonNullByDefault
  81 public class HomeAssistantThingHandler extends AbstractMQTTThingHandler
  82         implements ComponentDiscovered, Consumer<List<AbstractComponent<?>>> {
  83     public static final String AVAILABILITY_CHANNEL = "availability";
  84     private static final Comparator<AbstractComponent<?>> COMPONENT_COMPARATOR = Comparator
  85             .comparing((AbstractComponent<?> component) -> component.hasGroup())
  86             .thenComparing(AbstractComponent::getName);
  87     private static final URI UPDATABLE_CONFIG_DESCRIPTION_URI = URI.create("thing-type:mqtt:homeassistant-updatable");
  88
  89     private final Logger logger = LoggerFactory.getLogger(HomeAssistantThingHandler.class);
  90
  91     protected final MqttChannelTypeProvider channelTypeProvider;
  92     protected final MqttChannelStateDescriptionProvider stateDescriptionProvider;
  93     protected final ChannelTypeRegistry channelTypeRegistry;
  94     protected final Jinjava jinjava;
  95     public final int attributeReceiveTimeout;
  96     protected final DelayedBatchProcessing<AbstractComponent<?>> delayedProcessing;
  97     protected final DiscoverComponents discoverComponents;
  98
  99     private final Gson gson;
 100     protected final Map<@Nullable String, AbstractComponent<?>> haComponents = new HashMap<>();
 101     protected final Map<@Nullable String, AbstractComponent<?>> haComponentsByUniqueId = new HashMap<>();
 102     protected final Map<ChannelUID, ChannelState> channelStates = new HashMap<>();
 103
 104     protected HandlerConfiguration config = new HandlerConfiguration();
 105     private Set<HaID> discoveryHomeAssistantIDs = new HashSet<>();
 106
 107     private boolean started;
 108     private boolean newStyleChannels;
 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             MqttChannelStateDescriptionProvider stateDescriptionProvider, ChannelTypeRegistry channelTypeRegistry,
 122             Jinjava jinjava, int subscribeTimeout, int attributeReceiveTimeout) {
 123         super(thing, subscribeTimeout);
 124         this.gson = new GsonBuilder().registerTypeAdapterFactory(new ChannelConfigurationTypeAdapterFactory()).create();
 125         this.channelTypeProvider = channelTypeProvider;
 126         this.stateDescriptionProvider = stateDescriptionProvider;
 127         this.channelTypeRegistry = channelTypeRegistry;
 128         this.jinjava = jinjava;
 129         this.attributeReceiveTimeout = attributeReceiveTimeout;
 130         this.delayedProcessing = new DelayedBatchProcessing<>(attributeReceiveTimeout, this, scheduler);
 131
 132         newStyleChannels = "true".equals(thing.getProperties().get("newStyleChannels"));
 133
 134         this.discoverComponents = new DiscoverComponents(thing.getUID(), scheduler, this, this, gson, jinjava,
 135                 newStyleChannels);
 136     }
 137
 138     @Override
 139     public void initialize() {
 140         started = false;
 141
 142         config = getConfigAs(HandlerConfiguration.class);
 143         if (config.topics.isEmpty()) {
 144             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_ERROR, "Device topics unknown");
 145             return;
 146         }
 147         discoveryHomeAssistantIDs.addAll(HaID.fromConfig(config));
 148
 149         ThingTypeUID typeID = getThing().getThingTypeUID();
 150         for (Channel channel : thing.getChannels()) {
 151             final String groupID = channel.getUID().getGroupId();
 152             if (groupID != null) {
 153                 // Already restored component via another channel in the component?
 154                 AbstractComponent<?> component = haComponents.get(groupID);
 155                 if (component != null) {
 156                     continue;
 157                 }
 158             }
 159             Configuration channelConfig = channel.getConfiguration();
 160             if (!channelConfig.containsKey("component")
 161                     || !channelConfig.containsKey("objectid") | !channelConfig.containsKey("config")) {
 162                 // Must be a secondary channel
 163                 continue;
 164             }
 165
 166             HaID haID = HaID.fromConfig(config.basetopic, channelConfig);
 167
 168             if (!config.topics.contains(haID.getTopic())) {
 169                 // don't add a component for this channel that isn't configured on the thing
 170                 // anymore
 171                 // It will disappear from the thing when the thing type is updated below
 172                 continue;
 173             }
 174
 175             discoveryHomeAssistantIDs.add(haID);
 176             ThingUID thingUID = channel.getUID().getThingUID();
 177             String channelConfigurationJSON = (String) channelConfig.get("config");
 178             try {
 179                 AbstractComponent<?> component = ComponentFactory.createComponent(thingUID, haID,
 180                         channelConfigurationJSON, this, this, scheduler, gson, jinjava, newStyleChannels);
 181                 if (typeID.equals(MqttBindingConstants.HOMEASSISTANT_MQTT_THING)) {
 182                     typeID = calculateThingTypeUID(component);
 183                 }
 184
 185                 addComponent(component);
 186             } catch (ConfigurationException e) {
 187                 logger.warn("Cannot restore component {}: {}", thing, e.getMessage());
 188             }
 189         }
 190         if (updateThingType(typeID)) {
 191             super.initialize();
 192         }
 193     }
 194
 195     @Override
 196     public void dispose() {
 197         removeStateDescriptions();
 198         // super.dispose() calls stop()
 199         super.dispose();
 200     }
 201
 202     @Override
 203     public CompletableFuture<Void> unsubscribeAll() {
 204         // already unsubscribed everything by calling stop()
 205         return CompletableFuture.allOf();
 206     }
 207
 208     /**
 209      * Start a background discovery for the configured HA MQTT object-id.
 210      */
 211     @Override
 212     protected CompletableFuture<@Nullable Void> start(MqttBrokerConnection connection) {
 213         started = true;
 214
 215         connection.setQos(1);
 216         updateStatus(ThingStatus.UNKNOWN);
 217
 218         // Start all known components and channels within the components and put the Thing offline
 219         // if any subscribing failed ( == broker connection lost)
 220         CompletableFuture<@Nullable Void> future = CompletableFuture.allOf(super.start(connection),
 221                 haComponents.values().stream().map(e -> e.start(connection, scheduler, attributeReceiveTimeout))
 222                         .reduce(CompletableFuture.completedFuture(null), (a, v) -> a.thenCompose(b -> v)) // reduce to
 223                                                                                                           // one
 224                         .exceptionally(e -> {
 225                             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_ERROR, e.getMessage());
 226                             return null;
 227                         }));
 228
 229         return future
 230                 .thenCompose(b -> discoverComponents.startDiscovery(connection, 0, discoveryHomeAssistantIDs, this));
 231     }
 232
 233     @Override
 234     protected void stop() {
 235         if (started) {
 236             discoverComponents.stopDiscovery();
 237             delayedProcessing.join();
 238             // haComponents does not need to be synchronised -> the discovery thread is disabled
 239             haComponents.values().stream().map(AbstractComponent::stop) //
 240                     // we need to join all the stops, otherwise they might not be done when start is called
 241                     .collect(FutureCollector.allOf()).join();
 242
 243             started = false;
 244         }
 245         super.stop();
 246     }
 247
 248     @Override
 249     public @Nullable ChannelState getChannelState(ChannelUID channelUID) {
 250         synchronized (haComponents) { // sync whenever discoverComponents is started
 251             return channelStates.get(channelUID);
 252         }
 253     }
 254
 255     /**
 256      * Callback of {@link DiscoverComponents}. Add to a delayed batch processor.
 257      */
 258     @Override
 259     public void componentDiscovered(HaID homeAssistantTopicID, AbstractComponent<?> component) {
 260         delayedProcessing.accept(component);
 261     }
 262
 263     /**
 264      * Callback of {@link DelayedBatchProcessing}.
 265      * Add all newly discovered components to the Thing and start the components.
 266      */
 267     @Override
 268     public void accept(List<AbstractComponent<?>> discoveredComponentsList) {
 269         MqttBrokerConnection connection = this.connection;
 270         if (connection == null) {
 271             return;
 272         }
 273
 274         synchronized (haComponents) { // sync whenever discoverComponents is started
 275             ThingTypeUID typeID = getThing().getThingTypeUID();
 276             for (AbstractComponent<?> discovered : discoveredComponentsList) {
 277                 if (typeID.equals(MqttBindingConstants.HOMEASSISTANT_MQTT_THING)) {
 278                     typeID = calculateThingTypeUID(discovered);
 279                 }
 280                 AbstractComponent<?> known = haComponentsByUniqueId.get(discovered.getUniqueId());
 281                 // Is component already known?
 282                 if (known != null) {
 283                     if (discovered.getConfigHash() != known.getConfigHash()) {
 284                         // Don't wait for the future to complete. We are also not interested in failures.
 285                         // The component will be replaced in a moment.
 286                         known.stop();
 287                         haComponentsByUniqueId.remove(discovered.getUniqueId());
 288                         haComponents.remove(known.getComponentId());
 289                         if (!known.getComponentId().equals(discovered.getComponentId())) {
 290                             discovered.resolveConflict();
 291                         }
 292                     } else {
 293                         known.setConfigSeen();
 294                         continue;
 295                     }
 296                 }
 297
 298                 // Add component to the component map
 299                 addComponent(discovered);
 300                 // Start component / Subscribe to channel topics
 301                 discovered.start(connection, scheduler, 0).exceptionally(e -> {
 302                     logger.warn("Failed to start component {}", discovered.getHaID(), e);
 303                     return null;
 304                 });
 305
 306                 if (discovered instanceof Update) {
 307                     updateComponent = (Update) discovered;
 308                     updateComponent.setReleaseStateUpdateListener(this::releaseStateUpdated);
 309                 }
 310             }
 311             updateThingType(typeID);
 312         }
 313     }
 314
 315     @Override
 316     protected void updateThingStatus(boolean messageReceived, Optional<Boolean> availabilityTopicsSeen) {
 317         if (availabilityTopicsSeen.orElse(messageReceived)) {
 318             updateStatus(ThingStatus.ONLINE, ThingStatusDetail.NONE);
 319         } else {
 320             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.NONE);
 321         }
 322     }
 323
 324     @Override
 325     public void handleConfigurationUpdate(Map<String, Object> configurationParameters)
 326             throws ConfigValidationException {
 327         if (configurationParameters.containsKey("doUpdate")) {
 328             configurationParameters = new HashMap<>(configurationParameters);
 329             Object value = configurationParameters.remove("doUpdate");
 330             if (value instanceof Boolean doUpdate && doUpdate) {
 331                 Update updateComponent = this.updateComponent;
 332                 if (updateComponent == null) {
 333                     logger.warn(
 334                             "Received update command for Home Assistant device {}, but it does not have an update component.",
 335                             getThing().getUID());
 336                 } else {
 337                     updateComponent.doUpdate();
 338                 }
 339             }
 340         }
 341         super.handleConfigurationUpdate(configurationParameters);
 342     }
 343
 344     private boolean updateThingType(ThingTypeUID typeID) {
 345         // if this is a dynamic type, then we update the type
 346         if (!MqttBindingConstants.HOMEASSISTANT_MQTT_THING.equals(typeID)) {
 347             var thingTypeBuilder = channelTypeProvider.derive(typeID, MqttBindingConstants.HOMEASSISTANT_MQTT_THING);
 348
 349             if (getThing().getThingTypeUID().equals(MqttBindingConstants.HOMEASSISTANT_MQTT_THING)) {
 350                 logger.debug("Migrating Home Assistant thing {} from generic type to dynamic type {}",
 351                         getThing().getUID(), typeID);
 352
 353                 // just create an empty thing type for now; channel configurations won't follow over
 354                 // to the re-created Thing, so we need to re-discover them all anyway
 355                 channelTypeProvider.putThingType(thingTypeBuilder.build());
 356                 changeThingType(typeID, getConfig());
 357                 return false;
 358             }
 359
 360             synchronized (haComponents) { // sync whenever discoverComponents is started
 361                 var sortedComponents = haComponents.values().stream().sorted(COMPONENT_COMPARATOR).toList();
 362
 363                 var channelGroupTypes = sortedComponents.stream().map(c -> c.getChannelGroupType(typeID.getId()))
 364                         .filter(Objects::nonNull).map(Objects::requireNonNull).toList();
 365                 channelTypeProvider.updateChannelGroupTypesForPrefix(typeID.getId(), channelGroupTypes);
 366
 367                 var groupDefs = sortedComponents.stream().map(c -> c.getGroupDefinition(typeID.getId()))
 368                         .filter(Objects::nonNull).map(Objects::requireNonNull).toList();
 369                 var channelDefs = sortedComponents.stream().map(AbstractComponent::getChannelDefinitions)
 370                         .flatMap(List::stream).toList();
 371                 thingTypeBuilder.withChannelDefinitions(channelDefs).withChannelGroupDefinitions(groupDefs);
 372                 Update updateComponent = this.updateComponent;
 373                 if (updateComponent != null && updateComponent.isUpdatable()) {
 374                     thingTypeBuilder.withConfigDescriptionURI(UPDATABLE_CONFIG_DESCRIPTION_URI);
 375                 }
 376
 377                 channelTypeProvider.putThingType(thingTypeBuilder.build());
 378
 379                 removeStateDescriptions();
 380                 sortedComponents.stream().forEach(c -> c.addStateDescriptions(stateDescriptionProvider));
 381
 382                 ThingBuilder thingBuilder = editThing().withChannels();
 383
 384                 sortedComponents.stream().map(AbstractComponent::getChannels).flatMap(List::stream)
 385                         .forEach(c -> thingBuilder.withChannel(c));
 386
 387                 channelStates.clear();
 388                 sortedComponents.forEach(c -> c.getChannelStates(channelStates));
 389
 390                 updateThing(thingBuilder.build());
 391             }
 392         }
 393         return true;
 394     }
 395
 396     private ThingTypeUID calculateThingTypeUID(AbstractComponent component) {
 397         return new ThingTypeUID(MqttBindingConstants.BINDING_ID, MqttBindingConstants.HOMEASSISTANT_MQTT_THING.getId()
 398                 + "_" + component.getChannelConfiguration().getThingId(component.getHaID().objectID));
 399     }
 400
 401     @Override
 402     public void handleRemoval() {
 403         synchronized (haComponents) {
 404             channelTypeProvider.removeThingType(thing.getThingTypeUID());
 405             channelTypeProvider.removeChannelGroupTypesForPrefix(thing.getThingTypeUID().getId());
 406             removeStateDescriptions();
 407         }
 408         super.handleRemoval();
 409     }
 410
 411     private void removeStateDescriptions() {
 412         thing.getChannels().stream().forEach(c -> stateDescriptionProvider.remove(c.getUID()));
 413     }
 414
 415     private void releaseStateUpdated(Update.ReleaseState state) {
 416         Map<String, String> properties = editProperties();
 417         properties = state.appendToProperties(properties);
 418         updateProperties(properties);
 419     }
 420
 421     // should only be called when it's safe to access haComponents
 422     private void addComponent(AbstractComponent component) {
 423         AbstractComponent existing = haComponents.get(component.getComponentId());
 424         if (existing != null) {
 425             // rename the conflict
 426             haComponents.remove(existing.getComponentId());
 427             existing.resolveConflict();
 428             component.resolveConflict();
 429             haComponents.put(existing.getComponentId(), existing);
 430         }
 431         haComponents.put(component.getComponentId(), component);
 432         haComponentsByUniqueId.put(component.getUniqueId(), component);
 433     }
 434 }