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.mqtt.homeassistant.internal.handler;
16 import java.util.Comparator;
17 import java.util.HashMap;
18 import java.util.HashSet;
19 import java.util.List;
21 import java.util.Objects;
22 import java.util.Optional;
24 import java.util.concurrent.CompletableFuture;
25 import java.util.function.Consumer;
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.ComponentChannel;
37 import org.openhab.binding.mqtt.homeassistant.internal.DiscoverComponents;
38 import org.openhab.binding.mqtt.homeassistant.internal.DiscoverComponents.ComponentDiscovered;
39 import org.openhab.binding.mqtt.homeassistant.internal.HaID;
40 import org.openhab.binding.mqtt.homeassistant.internal.HandlerConfiguration;
41 import org.openhab.binding.mqtt.homeassistant.internal.component.AbstractComponent;
42 import org.openhab.binding.mqtt.homeassistant.internal.component.ComponentFactory;
43 import org.openhab.binding.mqtt.homeassistant.internal.component.Update;
44 import org.openhab.binding.mqtt.homeassistant.internal.config.ChannelConfigurationTypeAdapterFactory;
45 import org.openhab.binding.mqtt.homeassistant.internal.exception.ConfigurationException;
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;
60 import com.google.gson.Gson;
61 import com.google.gson.GsonBuilder;
62 import com.hubspot.jinjava.Jinjava;
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>
70 * The specification does not cover the case of disappearing Components. This handler doesn't as well therefore.<br>
73 * A Component Instance equals a Channel Group and the Component parts equal Channels.<br>
76 * If a Components configuration changes, the known ChannelGroupType and ChannelTypes are replaced with the new ones.
78 * @author David Graeff - Initial contribution
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");
89 private final Logger logger = LoggerFactory.getLogger(HomeAssistantThingHandler.class);
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;
99 private final Gson gson;
100 protected final Map<@Nullable String, AbstractComponent<?>> haComponents = new HashMap<>();
102 protected HandlerConfiguration config = new HandlerConfiguration();
103 private Set<HaID> discoveryHomeAssistantIDs = new HashSet<>();
105 private boolean started;
106 private boolean newStyleChannels;
107 private @Nullable Update updateComponent;
110 * Create a new thing handler for HomeAssistant MQTT components.
111 * A channel type provider and a topic value receive timeout must be provided.
113 * @param thing The thing of this handler
114 * @param channelTypeProvider A channel type provider
115 * @param subscribeTimeout Timeout for the entire tree parsing and subscription. In milliseconds.
116 * @param attributeReceiveTimeout The timeout per attribute field subscription. In milliseconds.
118 public HomeAssistantThingHandler(Thing thing, MqttChannelTypeProvider channelTypeProvider,
119 MqttChannelStateDescriptionProvider stateDescriptionProvider, ChannelTypeRegistry channelTypeRegistry,
120 Jinjava jinjava, int subscribeTimeout, int attributeReceiveTimeout) {
121 super(thing, subscribeTimeout);
122 this.gson = new GsonBuilder().registerTypeAdapterFactory(new ChannelConfigurationTypeAdapterFactory()).create();
123 this.channelTypeProvider = channelTypeProvider;
124 this.stateDescriptionProvider = stateDescriptionProvider;
125 this.channelTypeRegistry = channelTypeRegistry;
126 this.jinjava = jinjava;
127 this.attributeReceiveTimeout = attributeReceiveTimeout;
128 this.delayedProcessing = new DelayedBatchProcessing<>(attributeReceiveTimeout, this, scheduler);
130 newStyleChannels = "true".equals(thing.getProperties().get("newStyleChannels"));
132 this.discoverComponents = new DiscoverComponents(thing.getUID(), scheduler, this, this, gson, jinjava,
137 public void initialize() {
140 config = getConfigAs(HandlerConfiguration.class);
141 if (config.topics.isEmpty()) {
142 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_ERROR, "Device topics unknown");
145 discoveryHomeAssistantIDs.addAll(HaID.fromConfig(config));
147 ThingTypeUID typeID = getThing().getThingTypeUID();
148 for (Channel channel : thing.getChannels()) {
149 final String groupID = channel.getUID().getGroupId();
150 // Already restored component?
152 AbstractComponent<?> component = haComponents.get(groupID);
153 if (component != null) {
156 HaID haID = HaID.fromConfig(config.basetopic, channel.getConfiguration());
158 if (!config.topics.contains(haID.getTopic())) {
159 // don't add a component for this channel that isn't configured on the thing
161 // It will disappear from the thing when the thing type is updated below
165 discoveryHomeAssistantIDs.add(haID);
166 ThingUID thingUID = channel.getUID().getThingUID();
167 String channelConfigurationJSON = (String) channel.getConfiguration().get("config");
168 if (channelConfigurationJSON == null) {
169 logger.warn("Provided channel does not have a 'config' configuration key!");
172 component = ComponentFactory.createComponent(thingUID, haID, channelConfigurationJSON, this, this,
173 scheduler, gson, jinjava, newStyleChannels);
174 if (typeID.equals(MqttBindingConstants.HOMEASSISTANT_MQTT_THING)) {
175 typeID = calculateThingTypeUID(component);
178 haComponents.put(component.getGroupId(), component);
179 } catch (ConfigurationException e) {
180 logger.error("Cannot restore component {}: {}", thing, e.getMessage());
184 if (updateThingType(typeID)) {
190 public void dispose() {
191 removeStateDescriptions();
192 // super.dispose() calls stop()
197 public CompletableFuture<Void> unsubscribeAll() {
198 // already unsubscribed everything by calling stop()
199 return CompletableFuture.allOf();
203 * Start a background discovery for the configured HA MQTT object-id.
206 protected CompletableFuture<@Nullable Void> start(MqttBrokerConnection connection) {
209 connection.setQos(1);
210 updateStatus(ThingStatus.UNKNOWN);
212 // Start all known components and channels within the components and put the Thing offline
213 // if any subscribing failed ( == broker connection lost)
214 CompletableFuture<@Nullable Void> future = CompletableFuture.allOf(super.start(connection),
215 haComponents.values().stream().map(e -> e.start(connection, scheduler, attributeReceiveTimeout))
216 .reduce(CompletableFuture.completedFuture(null), (a, v) -> a.thenCompose(b -> v)) // reduce to
218 .exceptionally(e -> {
219 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_ERROR, e.getMessage());
224 .thenCompose(b -> discoverComponents.startDiscovery(connection, 0, discoveryHomeAssistantIDs, this));
228 protected void stop() {
230 discoverComponents.stopDiscovery();
231 delayedProcessing.join();
232 // haComponents does not need to be synchronised -> the discovery thread is disabled
233 haComponents.values().stream().map(AbstractComponent::stop) //
234 // we need to join all the stops, otherwise they might not be done when start is called
235 .collect(FutureCollector.allOf()).join();
243 public @Nullable ChannelState getChannelState(ChannelUID channelUID) {
245 if (channelUID.isInGroup()) {
246 componentId = channelUID.getGroupId();
248 componentId = channelUID.getId();
250 AbstractComponent<?> component;
251 synchronized (haComponents) { // sync whenever discoverComponents is started
252 component = haComponents.get(componentId);
254 if (component == null) {
255 component = haComponents.get("");
256 if (component == null) {
260 ComponentChannel componentChannel = component.getChannel(channelUID.getIdWithoutGroup());
261 if (componentChannel == null) {
264 return componentChannel.getState();
268 * Callback of {@link DiscoverComponents}. Add to a delayed batch processor.
271 public void componentDiscovered(HaID homeAssistantTopicID, AbstractComponent<?> component) {
272 delayedProcessing.accept(component);
276 * Callback of {@link DelayedBatchProcessing}.
277 * Add all newly discovered components to the Thing and start the components.
280 public void accept(List<AbstractComponent<?>> discoveredComponentsList) {
281 MqttBrokerConnection connection = this.connection;
282 if (connection == null) {
286 synchronized (haComponents) { // sync whenever discoverComponents is started
287 ThingTypeUID typeID = getThing().getThingTypeUID();
288 for (AbstractComponent<?> discovered : discoveredComponentsList) {
289 if (typeID.equals(MqttBindingConstants.HOMEASSISTANT_MQTT_THING)) {
290 typeID = calculateThingTypeUID(discovered);
292 String id = discovered.getGroupId();
293 AbstractComponent<?> known = haComponents.get(id);
294 // Is component already known?
296 if (discovered.getConfigHash() != known.getConfigHash()) {
297 // Don't wait for the future to complete. We are also not interested in failures.
298 // The component will be replaced in a moment.
301 known.setConfigSeen();
306 // Add component to the component map
307 haComponents.put(id, discovered);
308 // Start component / Subscribe to channel topics
309 discovered.start(connection, scheduler, 0).exceptionally(e -> {
310 logger.warn("Failed to start component {}", discovered.getHaID(), e);
314 if (discovered instanceof Update) {
315 updateComponent = (Update) discovered;
316 updateComponent.setReleaseStateUpdateListener(this::releaseStateUpdated);
319 updateThingType(typeID);
324 protected void updateThingStatus(boolean messageReceived, Optional<Boolean> availabilityTopicsSeen) {
325 if (availabilityTopicsSeen.orElse(messageReceived)) {
326 updateStatus(ThingStatus.ONLINE, ThingStatusDetail.NONE);
328 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.NONE);
333 public void handleConfigurationUpdate(Map<String, Object> configurationParameters)
334 throws ConfigValidationException {
335 if (configurationParameters.containsKey("doUpdate")) {
336 configurationParameters = new HashMap<>(configurationParameters);
337 Object value = configurationParameters.remove("doUpdate");
338 if (value instanceof Boolean doUpdate && doUpdate) {
339 Update updateComponent = this.updateComponent;
340 if (updateComponent == null) {
342 "Received update command for Home Assistant device {}, but it does not have an update component.",
343 getThing().getUID());
345 updateComponent.doUpdate();
349 super.handleConfigurationUpdate(configurationParameters);
352 private boolean updateThingType(ThingTypeUID typeID) {
353 // if this is a dynamic type, then we update the type
354 if (!MqttBindingConstants.HOMEASSISTANT_MQTT_THING.equals(typeID)) {
355 var thingTypeBuilder = channelTypeProvider.derive(typeID, MqttBindingConstants.HOMEASSISTANT_MQTT_THING);
357 if (getThing().getThingTypeUID().equals(MqttBindingConstants.HOMEASSISTANT_MQTT_THING)) {
358 logger.debug("Migrating Home Assistant thing {} from generic type to dynamic type {}",
359 getThing().getUID(), typeID);
361 // just create an empty thing type for now; channel configurations won't follow over
362 // to the re-created Thing, so we need to re-discover them all anyway
363 channelTypeProvider.putThingType(thingTypeBuilder.build());
364 changeThingType(typeID, getConfig());
368 synchronized (haComponents) { // sync whenever discoverComponents is started
369 var sortedComponents = haComponents.values().stream().sorted(COMPONENT_COMPARATOR).toList();
371 var channelGroupTypes = sortedComponents.stream().map(c -> c.getChannelGroupType(typeID.getId()))
372 .filter(Objects::nonNull).map(Objects::requireNonNull).toList();
373 channelTypeProvider.updateChannelGroupTypesForPrefix(typeID.getId(), channelGroupTypes);
375 var groupDefs = sortedComponents.stream().map(c -> c.getGroupDefinition(typeID.getId()))
376 .filter(Objects::nonNull).map(Objects::requireNonNull).toList();
377 var channelDefs = sortedComponents.stream().map(AbstractComponent::getChannelDefinitions)
378 .flatMap(List::stream).toList();
379 thingTypeBuilder.withChannelDefinitions(channelDefs).withChannelGroupDefinitions(groupDefs);
380 Update updateComponent = this.updateComponent;
381 if (updateComponent != null && updateComponent.isUpdatable()) {
382 thingTypeBuilder.withConfigDescriptionURI(UPDATABLE_CONFIG_DESCRIPTION_URI);
385 channelTypeProvider.putThingType(thingTypeBuilder.build());
387 removeStateDescriptions();
388 sortedComponents.stream().forEach(c -> c.addStateDescriptions(stateDescriptionProvider));
390 ThingBuilder thingBuilder = editThing().withChannels();
392 sortedComponents.stream().map(AbstractComponent::getChannels).flatMap(List::stream)
393 .forEach(c -> thingBuilder.withChannel(c));
395 updateThing(thingBuilder.build());
401 private ThingTypeUID calculateThingTypeUID(AbstractComponent component) {
402 return new ThingTypeUID(MqttBindingConstants.BINDING_ID, MqttBindingConstants.HOMEASSISTANT_MQTT_THING.getId()
403 + "_" + component.getChannelConfiguration().getThingId(component.getHaID().objectID));
407 public void handleRemoval() {
408 synchronized (haComponents) {
409 channelTypeProvider.removeThingType(thing.getThingTypeUID());
410 channelTypeProvider.removeChannelGroupTypesForPrefix(thing.getThingTypeUID().getId());
411 removeStateDescriptions();
413 super.handleRemoval();
416 private void removeStateDescriptions() {
417 thing.getChannels().stream().forEach(c -> stateDescriptionProvider.remove(c.getUID()));
420 private void releaseStateUpdated(Update.ReleaseState state) {
421 Map<String, String> properties = editProperties();
422 properties = state.appendToProperties(properties);
423 updateProperties(properties);