Refactor mqtt callbacks for update platform (#118131)
Jan Bouwhuis
GitHub
parent
e30297d837
commit
0f44ebd51e
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from functools import partial
|
||||
import logging
|
||||
from typing import Any, TypedDict, cast
|
||||
|
||||
|
|
@ -32,12 +33,7 @@ from .const import (
|
|||
CONF_STATE_TOPIC,
|
||||
PAYLOAD_EMPTY_JSON,
|
||||
)
|
||||
from .debug_info import log_messages
|
||||
from .mixins import (
|
||||
MqttEntity,
|
||||
async_setup_entity_entry_helper,
|
||||
write_state_on_attr_change,
|
||||
)
|
||||
from .mixins import MqttEntity, async_setup_entity_entry_helper
|
||||
from .models import MessageCallbackType, MqttValueTemplate, ReceiveMessage
|
||||
from .schemas import MQTT_ENTITY_COMMON_SCHEMA
|
||||
from .util import valid_publish_topic, valid_subscribe_topic
|
||||
|
|
@ -141,25 +137,104 @@ class MqttUpdate(MqttEntity, UpdateEntity, RestoreEntity):
|
|||
).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[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: _MqttUpdatePayloadType = {}
|
||||
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 = cast(_MqttUpdatePayloadType, 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 JSON_DECODE_EXCEPTIONS:
|
||||
_LOGGER.debug(
|
||||
(
|
||||
"No valid (JSON) payload detected after processing payload '%s'"
|
||||
" on topic %s"
|
||||
),
|
||||
payload,
|
||||
msg.topic,
|
||||
)
|
||||
json_payload["installed_version"] = str(payload)
|
||||
|
||||
if "installed_version" in json_payload:
|
||||
self._attr_installed_version = json_payload["installed_version"]
|
||||
|
||||
if "latest_version" in json_payload:
|
||||
self._attr_latest_version = json_payload["latest_version"]
|
||||
|
||||
if "title" in json_payload:
|
||||
self._attr_title = json_payload["title"]
|
||||
|
||||
if "release_summary" in json_payload:
|
||||
self._attr_release_summary = json_payload["release_summary"]
|
||||
|
||||
if "release_url" in json_payload:
|
||||
self._attr_release_url = json_payload["release_url"]
|
||||
|
||||
if "entity_picture" in json_payload:
|
||||
self._entity_picture = json_payload["entity_picture"]
|
||||
|
||||
@callback
|
||||
def _handle_latest_version_received(self, msg: ReceiveMessage) -> None:
|
||||
"""Handle receiving latest version via MQTT."""
|
||||
latest_version = self._templates[CONF_LATEST_VERSION_TEMPLATE](msg.payload)
|
||||
|
||||
if isinstance(latest_version, str) and latest_version != "":
|
||||
self._attr_latest_version = latest_version
|
||||
|
||||
def _prepare_subscribe_topics(self) -> None:
|
||||
"""(Re)Subscribe to topics."""
|
||||
topics: dict[str, Any] = {}
|
||||
|
||||
def add_subscription(
|
||||
topics: dict[str, Any], topic: str, msg_callback: MessageCallbackType
|
||||
topics: dict[str, Any],
|
||||
topic: str,
|
||||
msg_callback: MessageCallbackType,
|
||||
tracked_attributes: set[str],
|
||||
) -> None:
|
||||
if self._config.get(topic) is not None:
|
||||
topics[topic] = {
|
||||
"topic": self._config[topic],
|
||||
"msg_callback": msg_callback,
|
||||
"msg_callback": partial(
|
||||
self._message_callback, msg_callback, tracked_attributes
|
||||
),
|
||||
"entity_id": self.entity_id,
|
||||
"qos": self._config[CONF_QOS],
|
||||
"encoding": self._config[CONF_ENCODING] or None,
|
||||
}
|
||||
|
||||
@callback
|
||||
@log_messages(self.hass, self.entity_id)
|
||||
@write_state_on_attr_change(
|
||||
self,
|
||||
add_subscription(
|
||||
topics,
|
||||
CONF_STATE_TOPIC,
|
||||
self._handle_state_message_received,
|
||||
{
|
||||
"_attr_installed_version",
|
||||
"_attr_latest_version",
|
||||
|
|
@ -169,84 +244,11 @@ class MqttUpdate(MqttEntity, UpdateEntity, RestoreEntity):
|
|||
"_entity_picture",
|
||||
},
|
||||
)
|
||||
def handle_state_message_received(msg: ReceiveMessage) -> None:
|
||||
"""Handle receiving state message via MQTT."""
|
||||
payload = self._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: _MqttUpdatePayloadType = {}
|
||||
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 = cast(_MqttUpdatePayloadType, 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 JSON_DECODE_EXCEPTIONS:
|
||||
_LOGGER.debug(
|
||||
(
|
||||
"No valid (JSON) payload detected after processing payload '%s'"
|
||||
" on topic %s"
|
||||
),
|
||||
payload,
|
||||
msg.topic,
|
||||
)
|
||||
json_payload["installed_version"] = str(payload)
|
||||
|
||||
if "installed_version" in json_payload:
|
||||
self._attr_installed_version = json_payload["installed_version"]
|
||||
|
||||
if "latest_version" in json_payload:
|
||||
self._attr_latest_version = json_payload["latest_version"]
|
||||
|
||||
if "title" in json_payload:
|
||||
self._attr_title = json_payload["title"]
|
||||
|
||||
if "release_summary" in json_payload:
|
||||
self._attr_release_summary = json_payload["release_summary"]
|
||||
|
||||
if "release_url" in json_payload:
|
||||
self._attr_release_url = json_payload["release_url"]
|
||||
|
||||
if "entity_picture" in json_payload:
|
||||
self._entity_picture = json_payload["entity_picture"]
|
||||
|
||||
add_subscription(topics, CONF_STATE_TOPIC, handle_state_message_received)
|
||||
|
||||
@callback
|
||||
@log_messages(self.hass, self.entity_id)
|
||||
@write_state_on_attr_change(self, {"_attr_latest_version"})
|
||||
def handle_latest_version_received(msg: ReceiveMessage) -> None:
|
||||
"""Handle receiving latest version via MQTT."""
|
||||
latest_version = self._templates[CONF_LATEST_VERSION_TEMPLATE](msg.payload)
|
||||
|
||||
if isinstance(latest_version, str) and latest_version != "":
|
||||
self._attr_latest_version = latest_version
|
||||
|
||||
add_subscription(
|
||||
topics, CONF_LATEST_VERSION_TOPIC, handle_latest_version_received
|
||||
topics,
|
||||
CONF_LATEST_VERSION_TOPIC,
|
||||
self._handle_latest_version_received,
|
||||
{"_attr_latest_version"},
|
||||
)
|
||||
|
||||
self._sub_state = subscription.async_prepare_subscribe_topics(
|
||||
|
|
|
|||
Loading…
Reference in New Issue