Documentation Enhancement: A working MQTT example app

[rct]

trafficstars

Is there an existing feature request for this?

• [X] I have searched the existing issues

Your feature request

Unless I've missed something, there isn't an example of a working MQTT app. I don't see any MQTT examples in: https://github.com/AppDaemon/appdaemon/tree/dev/conf/example_apps

There are snippets in the documentation, but not the full picture of what is needed in one place to have a working example. If one is considering using AppDaemon for the first time, it might dissuade them from choosing AppDaemon.

Based on some of the questions I've seen when searching, a good example MQTT AppDaemon app would:

• Subscribe to multiple (sub)topics or a wildcard topic
  • Be able to distinguish which topic a message was received for
• Maintain an internal state for a device that has subtopics: .../temperaure, .../humidity, .../battery, etc.
• Do "something"
  • Calculate some stats: high, low, average, message counts, etc. and publish it back to MQTT.
  • publish a single JSON state record with all the subtopics in one record.

[markfickett]

I wasn't able to get anything working deriving from mqtt.Mqtt as in the snippet in the MQTT API reference. So I would have definitely found a working example in the docs very useful.

Following some examples in the HomeAssistant community forums, I came up with this working app. It doesn't cover all of what rct suggests, but could potentially be adjusted to be a useful example in the docs:

import json
from typing import Dict

import hassapi as hass


_MQTT_PLUGIN_NAMESPACE = "mqtt"  # matches appdeamon.yaml


class RemoteTemp(hass.Hass):
  """Forward a AM2301 temperature reading to a Tasmota MiElHVAC.

  This allows using a remote temperature sensor with a heat pump.
  """

  def initialize(self) -> None:
    self._mqtt = self.get_plugin_api("MQTT")
    self._sensor_topic = self.args["sensor_topic"]
    self._command_topic = self.args["command_topic"]
    self._mqtt.mqtt_subscribe(self._sensor_topic, namespace=_MQTT_PLUGIN_NAMESPACE)
    self._mqtt.listen_event( self._temperature_received, "MQTT_MESSAGE", topic=self._sensor_topic)

  def _temperature_received(
    self, event: str, data: Dict, unused_kwargs: Dict
  ) -> None:
    # Example MQTT message:
    # tele/tasmota_884477/SENSOR {"Time":"2023-11-09T22:11:05","Switch1":"OFF","Switch2":"OFF","Switch3":"OFF","Switch4":"OFF","AM2301":{"Temperature":22.2,"Humidity":35.5,"DewPoint":6.2},"TempUnit":"C"}
    payload = json.loads(data["payload"])
    temperature_c = payload["AM2301"]["Temperature"]
    unit = payload["TempUnit"]
    if unit != "C":
      self.log(f"Temperature from {self._sensor_topic} has unit {unit}, not C.")
    self._mqtt.mqtt_publish(
      self._command_topic,
      # Round the temperature value since the MiElHVAC Tasmota driver truncates
      # to an integer value.
      payload=str(round(temperature_c)),
      namespace=_MQTT_PLUGIN_NAMESPACE,
    )