Log transmitted MQTT messages

homeassistant/components/mqtt/__init__.py

@@ -1184,7 +1184,7 @@ def _matcher_for_topic(subscription: str) -> Any:
 async def websocket_mqtt_info(hass, connection, msg):
     """Get MQTT debug info for device."""
     device_id = msg["device_id"]
-    mqtt_info = await debug_info.info_for_device(hass, device_id)
+    mqtt_info = debug_info.info_for_device(hass, device_id)
 
     connection.send_result(msg["id"], mqtt_info)
 

homeassistant/components/mqtt/alarm_control_panel.py

@@ -328,8 +328,7 @@ async def _publish(self, code, action):
         """Publish via mqtt."""
         variables = {"action": action, "code": code}
         payload = self._command_template(None, variables=variables)
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._config[CONF_COMMAND_TOPIC],
             payload,
             self._config[CONF_QOS],

homeassistant/components/mqtt/button.py

@@ -112,8 +112,7 @@ async def async_press(self, **kwargs):
         This method is a coroutine.
         """
         payload = self._command_template(self._config[CONF_PAYLOAD_PRESS])
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._config[CONF_COMMAND_TOPIC],
             payload,
             self._config[CONF_QOS],

homeassistant/components/mqtt/climate.py

@@ -708,8 +708,7 @@ def fan_modes(self):
 
     async def _publish(self, topic, payload):
         if self._topic[topic] is not None:
-            await mqtt.async_publish(
-                self.hass,
+            await self.async_publish(
                 self._topic[topic],
                 payload,
                 self._config[CONF_QOS],

homeassistant/components/mqtt/cover.py

@@ -534,8 +534,7 @@ async def async_open_cover(self, **kwargs):
 
         This method is a coroutine.
         """
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._config.get(CONF_COMMAND_TOPIC),
             self._config[CONF_PAYLOAD_OPEN],
             self._config[CONF_QOS],
@@ -556,8 +555,7 @@ async def async_close_cover(self, **kwargs):
 
         This method is a coroutine.
         """
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._config.get(CONF_COMMAND_TOPIC),
             self._config[CONF_PAYLOAD_CLOSE],
             self._config[CONF_QOS],
@@ -578,8 +576,7 @@ async def async_stop_cover(self, **kwargs):
 
         This method is a coroutine.
         """
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._config.get(CONF_COMMAND_TOPIC),
             self._config[CONF_PAYLOAD_STOP],
             self._config[CONF_QOS],
@@ -599,8 +596,7 @@ async def async_open_cover_tilt(self, **kwargs):
             "tilt_max": self._config.get(CONF_TILT_MAX),
         }
         tilt_payload = self._set_tilt_template(tilt_open_position, variables=variables)
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._config.get(CONF_TILT_COMMAND_TOPIC),
             tilt_payload,
             self._config[CONF_QOS],
@@ -627,8 +623,7 @@ async def async_close_cover_tilt(self, **kwargs):
         tilt_payload = self._set_tilt_template(
             tilt_closed_position, variables=variables
         )
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._config.get(CONF_TILT_COMMAND_TOPIC),
             tilt_payload,
             self._config[CONF_QOS],
@@ -657,8 +652,7 @@ async def async_set_cover_tilt_position(self, **kwargs):
         }
         tilt = self._set_tilt_template(tilt, variables=variables)
 
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._config.get(CONF_TILT_COMMAND_TOPIC),
             tilt,
             self._config[CONF_QOS],
@@ -685,8 +679,7 @@ async def async_set_cover_position(self, **kwargs):
         }
         position = self._set_position_template(position, variables=variables)
 
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._config.get(CONF_SET_POSITION_TOPIC),
             position,
             self._config[CONF_QOS],

homeassistant/components/mqtt/debug_info.py

@@ -3,13 +3,18 @@
 
 from collections import deque
 from collections.abc import Callable
+import datetime as dt
 from functools import wraps
 from typing import Any
 
+import attr
+
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+from homeassistant.util import dt as dt_util
 
 from .const import ATTR_DISCOVERY_PAYLOAD, ATTR_DISCOVERY_TOPIC
-from .models import MessageCallbackType
+from .models import MessageCallbackType, PublishPayloadType
 
 DATA_MQTT_DEBUG_INFO = "mqtt_debug_info"
 STORED_MESSAGES = 10
@@ -42,14 +47,50 @@ def wrapper(msg: Any) -> None:
     return _decorator
 
 
+@attr.s(slots=True, frozen=True)
+class TimestampedPublishMessage:
+    """MQTT Message."""
+
+    topic: str = attr.ib()
+    payload: PublishPayloadType = attr.ib()
+    qos: int = attr.ib()
+    retain: bool = attr.ib()
+    timestamp: dt.datetime = attr.ib(default=None)
+
+
+def log_message(
+    hass: HomeAssistant,
+    entity_id: str,
+    topic: str,
+    payload: PublishPayloadType,
+    qos: int,
+    retain: bool,
+) -> None:
+    """Log an outgoing MQTT message."""
+    debug_info = hass.data.setdefault(
+        DATA_MQTT_DEBUG_INFO, {"entities": {}, "triggers": {}}
+    )
+    entity_info = debug_info["entities"].setdefault(
+        entity_id, {"subscriptions": {}, "discovery_data": {}, "transmitted": {}}
+    )
+    if topic not in entity_info["transmitted"]:
+        entity_info["transmitted"][topic] = {
+            "messages": deque([], STORED_MESSAGES),
+        }
+    msg = TimestampedPublishMessage(
+        topic, payload, qos, retain, timestamp=dt_util.utcnow()
+    )
+    entity_info["transmitted"][topic]["messages"].append(msg)
+
+
 def add_subscription(hass, message_callback, subscription):
     """Prepare debug data for subscription."""
     if entity_id := getattr(message_callback, "__entity_id", None):
         debug_info = hass.data.setdefault(
             DATA_MQTT_DEBUG_INFO, {"entities": {}, "triggers": {}}
         )
         entity_info = debug_info["entities"].setdefault(
-            entity_id, {"subscriptions": {}, "discovery_data": {}}
+            entity_id, {"subscriptions": {}, "discovery_data": {}, "transmitted": {}}
         )
         if subscription not in entity_info["subscriptions"]:
             entity_info["subscriptions"][subscription] = {
@@ -80,7 +121,7 @@ def add_entity_discovery_data(hass, discovery_data, entity_id):
         DATA_MQTT_DEBUG_INFO, {"entities": {}, "triggers": {}}
     )
     entity_info = debug_info["entities"].setdefault(
-        entity_id, {"subscriptions": {}, "discovery_data": {}}
+        entity_id, {"subscriptions": {}, "discovery_data": {}, "transmitted": {}}
     )
     entity_info["discovery_data"] = discovery_data
 
@@ -118,10 +159,10 @@ def remove_trigger_discovery_data(hass, discovery_hash):
     hass.data[DATA_MQTT_DEBUG_INFO]["triggers"][discovery_hash]["discovery_data"] = None
 
 
-async def info_for_device(hass, device_id):
+def info_for_device(hass, device_id):
     """Get debug info for a device."""
     mqtt_info = {"entities": [], "triggers": []}
-    entity_registry = await hass.helpers.entity_registry.async_get_registry()
+    entity_registry = er.async_get(hass)
 
     entries = hass.helpers.entity_registry.async_entries_for_device(
         entity_registry, device_id, include_disabled_entities=True
@@ -150,6 +191,22 @@ async def info_for_device(hass, device_id):
             }
             for topic, subscription in entity_info["subscriptions"].items()
         ]
+        transmitted = [
+            {
+                "topic": topic,
+                "messages": [
+                    {
+                        "payload": str(msg.payload),
+                        "qos": msg.qos,
+                        "retain": msg.retain,
+                        "time": msg.timestamp,
+                        "topic": msg.topic,
+                    }
+                    for msg in list(subscription["messages"])
+                ],
+            }
+            for topic, subscription in entity_info["transmitted"].items()
+        ]
         discovery_data = {
             "topic": entity_info["discovery_data"].get(ATTR_DISCOVERY_TOPIC, ""),
             "payload": entity_info["discovery_data"].get(ATTR_DISCOVERY_PAYLOAD, ""),
@@ -159,6 +216,7 @@ async def info_for_device(hass, device_id):
                 "entity_id": entry.entity_id,
                 "subscriptions": subscriptions,
                 "discovery_data": discovery_data,
+                "transmitted": transmitted,
             }
         )
 

homeassistant/components/mqtt/fan.py

@@ -540,8 +540,7 @@ async def async_turn_on(
         This method is a coroutine.
         """
         mqtt_payload = self._command_templates[CONF_STATE](self._payload["STATE_ON"])
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_COMMAND_TOPIC],
             mqtt_payload,
             self._config[CONF_QOS],
@@ -562,8 +561,7 @@ async def async_turn_off(self, **kwargs) -> None:
         This method is a coroutine.
         """
         mqtt_payload = self._command_templates[CONF_STATE](self._payload["STATE_OFF"])
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_COMMAND_TOPIC],
             mqtt_payload,
             self._config[CONF_QOS],
@@ -583,8 +581,7 @@ async def async_set_percentage(self, percentage: int) -> None:
             percentage_to_ranged_value(self._speed_range, percentage)
         )
         mqtt_payload = self._command_templates[ATTR_PERCENTAGE](percentage_payload)
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_PERCENTAGE_COMMAND_TOPIC],
             mqtt_payload,
             self._config[CONF_QOS],
@@ -607,8 +604,7 @@ async def async_set_preset_mode(self, preset_mode: str) -> None:
 
         mqtt_payload = self._command_templates[ATTR_PRESET_MODE](preset_mode)
 
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_PRESET_MODE_COMMAND_TOPIC],
             mqtt_payload,
             self._config[CONF_QOS],
@@ -634,8 +630,7 @@ async def async_oscillate(self, oscillating: bool) -> None:
                 self._payload["OSCILLATE_OFF_PAYLOAD"]
             )
 
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_OSCILLATION_COMMAND_TOPIC],
             mqtt_payload,
             self._config[CONF_QOS],

homeassistant/components/mqtt/humidifier.py

@@ -415,8 +415,7 @@ async def async_turn_on(
         This method is a coroutine.
         """
         mqtt_payload = self._command_templates[CONF_STATE](self._payload["STATE_ON"])
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_COMMAND_TOPIC],
             mqtt_payload,
             self._config[CONF_QOS],
@@ -433,8 +432,7 @@ async def async_turn_off(self, **kwargs) -> None:
         This method is a coroutine.
         """
         mqtt_payload = self._command_templates[CONF_STATE](self._payload["STATE_OFF"])
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_COMMAND_TOPIC],
             mqtt_payload,
             self._config[CONF_QOS],
@@ -451,8 +449,7 @@ async def async_set_humidity(self, humidity: int) -> None:
         This method is a coroutine.
         """
         mqtt_payload = self._command_templates[ATTR_HUMIDITY](humidity)
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_TARGET_HUMIDITY_COMMAND_TOPIC],
             mqtt_payload,
             self._config[CONF_QOS],
@@ -475,8 +472,7 @@ async def async_set_mode(self, mode: str) -> None:
 
         mqtt_payload = self._command_templates[ATTR_MODE](mode)
 
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_MODE_COMMAND_TOPIC],
             mqtt_payload,
             self._config[CONF_QOS],

homeassistant/components/mqtt/light/schema_basic.py

@@ -831,8 +831,7 @@ async def async_turn_on(self, **kwargs):  # noqa: C901
 
         async def publish(topic, payload):
             """Publish an MQTT message."""
-            await mqtt.async_publish(
-                self.hass,
+            await self.async_publish(
                 self._topic[topic],
                 payload,
                 self._config[CONF_QOS],
@@ -1079,8 +1078,7 @@ async def async_turn_off(self, **kwargs):
 
         This method is a coroutine.
         """
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_COMMAND_TOPIC],
             self._payload["off"],
             self._config[CONF_QOS],

homeassistant/components/mqtt/light/schema_json.py

@@ -634,8 +634,7 @@ async def async_turn_on(self, **kwargs):  # noqa: C901
                 self._white_value = kwargs[ATTR_WHITE_VALUE]
                 should_update = True
 
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_COMMAND_TOPIC],
             json.dumps(message),
             self._config[CONF_QOS],
@@ -660,8 +659,7 @@ async def async_turn_off(self, **kwargs):
 
         self._set_flash_and_transition(message, **kwargs)
 
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topic[CONF_COMMAND_TOPIC],
             json.dumps(message),
             self._config[CONF_QOS],

homeassistant/components/mqtt/light/schema_template.py

@@ -388,8 +388,7 @@ async def async_turn_on(self, **kwargs):
         if ATTR_TRANSITION in kwargs:
             values["transition"] = kwargs[ATTR_TRANSITION]
 
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topics[CONF_COMMAND_TOPIC],
             self._templates[CONF_COMMAND_ON_TEMPLATE].async_render(
                 parse_result=False, **values
@@ -414,8 +413,7 @@ async def async_turn_off(self, **kwargs):
         if ATTR_TRANSITION in kwargs:
             values["transition"] = kwargs[ATTR_TRANSITION]
 
-        await mqtt.async_publish(
-            self.hass,
+        await self.async_publish(
             self._topics[CONF_COMMAND_TOPIC],
             self._templates[CONF_COMMAND_OFF_TEMPLATE].async_render(
                 parse_result=False, **values