update.py

Go to the documentation of this file.

"""Configure update platform in a device through MQTT topic."""
 
from __future__ import annotations
 
import logging
from typing import Any
 
import voluptuous as vol
 
from homeassistant.components import update
from homeassistant.components.update import (
    DEVICE_CLASSES_SCHEMA,
    UpdateEntity,
    UpdateEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DEVICE_CLASS, CONF_NAME, CONF_VALUE_TEMPLATE
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType, VolSchemaType
from homeassistant.util.json import JSON_DECODE_EXCEPTIONS, json_loads
 
from . import subscription
from .config import DEFAULT_RETAIN, MQTT_RO_SCHEMA
from .const import CONF_COMMAND_TOPIC, CONF_RETAIN, CONF_STATE_TOPIC, PAYLOAD_EMPTY_JSON
from .entity import MqttEntity, async_setup_entity_entry_helper
from .models import MqttValueTemplate, ReceiveMessage
from .schemas import MQTT_ENTITY_COMMON_SCHEMA
from .util import valid_publish_topic, valid_subscribe_topic
 
_LOGGER = logging.getLogger(__name__)
 
PARALLEL_UPDATES = 0
 
DEFAULT_NAME = "MQTT Update"
 
CONF_DISPLAY_PRECISION = "display_precision"
CONF_LATEST_VERSION_TEMPLATE = "latest_version_template"
CONF_LATEST_VERSION_TOPIC = "latest_version_topic"
CONF_PAYLOAD_INSTALL = "payload_install"
CONF_RELEASE_SUMMARY = "release_summary"
CONF_RELEASE_URL = "release_url"
CONF_TITLE = "title"
 
 
PLATFORM_SCHEMA_MODERN = MQTT_RO_SCHEMA.extend(
    {
        vol.Optional(CONF_COMMAND_TOPIC): valid_publish_topic,
        vol.Optional(CONF_DEVICE_CLASS): vol.Any(DEVICE_CLASSES_SCHEMA, None),
        vol.Optional(CONF_DISPLAY_PRECISION, default=0): cv.positive_int,
        vol.Optional(CONF_LATEST_VERSION_TEMPLATE): cv.template,
        vol.Optional(CONF_LATEST_VERSION_TOPIC): valid_subscribe_topic,
        vol.Optional(CONF_NAME): vol.Any(cv.string, None),
        vol.Optional(CONF_PAYLOAD_INSTALL): cv.string,
        vol.Optional(CONF_RELEASE_SUMMARY): cv.string,
        vol.Optional(CONF_RELEASE_URL): cv.string,
        vol.Optional(CONF_RETAIN, default=DEFAULT_RETAIN): cv.boolean,
        vol.Optional(CONF_TITLE): cv.string,
    },
).extend(MQTT_ENTITY_COMMON_SCHEMA.schema)
 
 
DISCOVERY_SCHEMA = vol.All(PLATFORM_SCHEMA_MODERN.extend({}, extra=vol.REMOVE_EXTRA))
 
 
MQTT_JSON_UPDATE_SCHEMA = vol.Schema(
    {
        vol.Optional("installed_version"): cv.string,
        vol.Optional("latest_version"): cv.string,
        vol.Optional("title"): cv.string,
        vol.Optional("release_summary"): cv.string,
        vol.Optional("release_url"): cv.url,
        vol.Optional("entity_picture"): cv.url,
        vol.Optional("in_progress"): cv.boolean,
        vol.Optional("update_percentage"): vol.Any(vol.Range(min=0, max=100), None),
    }
)
 
 
async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up MQTT update entity through YAML and through MQTT discovery."""
    async_setup_entity_entry_helper(
        hass,
        config_entry,
        MqttUpdate,
        update.DOMAIN,
        async_add_entities,
        DISCOVERY_SCHEMA,
        PLATFORM_SCHEMA_MODERN,
    )
 
 
class MqttUpdate(MqttEntity, UpdateEntity, RestoreEntity):
    """Representation of the MQTT update entity."""
 
    _default_name = DEFAULT_NAME
    _entity_id_format = update.ENTITY_ID_FORMAT
 
    @property
    def entity_picture(self) -> str | None:
        """Return the entity picture to use in the frontend."""
        if self._attr_entity_picture_attr_entity_picture_attr_entity_picture is not None:
            return self._attr_entity_picture_attr_entity_picture_attr_entity_picture
 
        return super().entity_picture
 
    @staticmethod
    def config_schema() -> VolSchemaType:
        """Return the config schema."""
        return DISCOVERY_SCHEMA
 
    def _setup_from_config(self, config: ConfigType) -> None:
        """(Re)Setup the entity."""
        self._attr_device_class_attr_device_class = self._config_config.get(CONF_DEVICE_CLASS)
        self._attr_display_precision_attr_display_precision = self._config_config[CONF_DISPLAY_PRECISION]
        self._attr_release_summary_attr_release_summary = self._config_config.get(CONF_RELEASE_SUMMARY)
        self._attr_release_url_attr_release_url = self._config_config.get(CONF_RELEASE_URL)
        self._attr_title_attr_title = self._config_config.get(CONF_TITLE)
        self._templates_templates = {
            CONF_VALUE_TEMPLATE: MqttValueTemplate(
                config.get(CONF_VALUE_TEMPLATE),
                entity=self,
            ).async_render_with_possible_json_value,
            CONF_LATEST_VERSION_TEMPLATE: MqttValueTemplate(
                config.get(CONF_LATEST_VERSION_TEMPLATE),
                entity=self,
            ).async_render_with_possible_json_value,
        }
 
    @callback
    def _handle_state_message_received(self, msg: ReceiveMessage) -> None:
        """Handle receiving state message via MQTT."""
        payload = self._templates_templates[CONF_VALUE_TEMPLATE](msg.payload)
 
        if not payload or payload == PAYLOAD_EMPTY_JSON:
            _LOGGER.debug(
                "Ignoring empty payload '%s' after rendering for topic %s",
                payload,
                msg.topic,
            )
            return
 
        json_payload: dict[str, Any] = {}
        try:
            rendered_json_payload = json_loads(payload)
            if isinstance(rendered_json_payload, dict):
                _LOGGER.debug(
                    (
                        "JSON payload detected after processing payload '%s' on"
                        " topic %s"
                    ),
                    rendered_json_payload,
                    msg.topic,
                )
                json_payload = MQTT_JSON_UPDATE_SCHEMA(rendered_json_payload)
            else:
                _LOGGER.debug(
                    (
                        "Non-dictionary JSON payload detected after processing"
                        " payload '%s' on topic %s"
                    ),
                    payload,
                    msg.topic,
                )
                json_payload = {"installed_version": str(payload)}
        except vol.MultipleInvalid as exc:
            _LOGGER.warning(
                (
                    "Schema violation after processing payload '%s'"
                    " on topic '%s' for entity '%s': %s"
                ),
                payload,
                msg.topic,
                self.entity_identity_id,
                exc,
            )
            return
        except JSON_DECODE_EXCEPTIONS:
            _LOGGER.debug(
                (
                    "No valid (JSON) payload detected after processing payload '%s'"
                    " on topic '%s' for entity '%s'"
                ),
                payload,
                msg.topic,
                self.entity_identity_id,
            )
            json_payload["installed_version"] = str(payload)
 
        if "installed_version" in json_payload:
            self._attr_installed_version_attr_installed_version = json_payload["installed_version"]
 
        if "latest_version" in json_payload:
            self._attr_latest_version_attr_latest_version = json_payload["latest_version"]
 
        if "title" in json_payload:
            self._attr_title_attr_title = json_payload["title"]
 
        if "release_summary" in json_payload:
            self._attr_release_summary_attr_release_summary = json_payload["release_summary"]
 
        if "release_url" in json_payload:
            self._attr_release_url_attr_release_url = json_payload["release_url"]
 
        if "entity_picture" in json_payload:
            self._attr_entity_picture_attr_entity_picture_attr_entity_picture = json_payload["entity_picture"]
 
        if "update_percentage" in json_payload:
            self._attr_update_percentage_attr_update_percentage = json_payload["update_percentage"]
            self._attr_in_progress_attr_in_progress_attr_in_progress = self._attr_update_percentage_attr_update_percentage is not None
 
        if "in_progress" in json_payload:
            self._attr_in_progress_attr_in_progress_attr_in_progress = json_payload["in_progress"]
 
    @callback
    def _handle_latest_version_received(self, msg: ReceiveMessage) -> None:
        """Handle receiving latest version via MQTT."""
        latest_version = self._templates_templates[CONF_LATEST_VERSION_TEMPLATE](msg.payload)
 
        if isinstance(latest_version, str) and latest_version != "":
            self._attr_latest_version_attr_latest_version = latest_version
 
    @callback
    def _prepare_subscribe_topics(self) -> None:
        """(Re)Subscribe to topics."""
        self.add_subscriptionadd_subscription(
            CONF_STATE_TOPIC,
            self._handle_state_message_received_handle_state_message_received,
            {
                "_attr_entity_picture",
                "_attr_in_progress",
                "_attr_installed_version",
                "_attr_latest_version",
                "_attr_title",
                "_attr_release_summary",
                "_attr_release_url",
                "_attr_update_percentage",
            },
        )
        self.add_subscriptionadd_subscription(
            CONF_LATEST_VERSION_TOPIC,
            self._handle_latest_version_received_handle_latest_version_received,
            {"_attr_latest_version"},
        )
 
    async def _subscribe_topics(self) -> None:
        """(Re)Subscribe to topics."""
        subscription.async_subscribe_topics_internal(self.hasshasshass, self._sub_state_sub_state)
 
    async def async_install(
        self, version: str | None, backup: bool, **kwargs: Any
    ) -> None:
        """Update the current value."""
        payload = self._config_config[CONF_PAYLOAD_INSTALL]
        await self.async_publish_with_configasync_publish_with_config(self._config_config[CONF_COMMAND_TOPIC], payload)
 
    @property
    def supported_features(self) -> UpdateEntityFeature:
        """Return the list of supported features."""
        support = UpdateEntityFeature(UpdateEntityFeature.PROGRESS)
 
        if self._config_config.get(CONF_COMMAND_TOPIC) is not None:
            support |= UpdateEntityFeature.INSTALL
 
        return support