Documentation Enhancement: A working MQTT example app

[rct]

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,
    )

[canassa]

I got here by searching on Google on how to get the MQTT working. THANK YOU @markfickett and @rct! Your example solved my issue

[rct]

I had forgotten about this issue. I did get a number of the things that I wanted to do accomplished. I never posted anything because I wound up with a bunch of trial and error in terms of the code and the script configuration in order to get subscriptions that did what I wanted. I wasn't sure how much was a problem with my understanding versus bugs or design issues.

IIRC trying to configure the connection and the subscription in the script config had some unexpected side effects.

I was thinking about AppDaemon MQTT recently and wondering whether it would be possible to manage connections to two different MQTT servers in order to selectively parse and forward messages that meet certain criteria.