command_endpoints: initial MQTT command endpoint

As there is no support for this in the UI yet:

curl http://your.was.host:8502/api/config?type=config | jq > was-config.json

Modify was-config.json to include:

  "command_endpoint": "MQTT",
  "mqtt_auth_type": "userpw",
  "mqtt_host": "your.mqtt.host",
  "mqtt_password": "bar",
  "mqtt_port": 1883,
  "mqtt_tls": false,
  "mqtt_topic": "foo",
  "mqtt_username: "foo",

Then:

curl -X POST -H "content-type: application/json" -d @was-config.json  "http://your.was.host:502/api/config?type=config&apply=false"

Finally in WAS UI apply config to the client you want to test the MQTT
endpoint with.

api.py

@@ -32,6 +32,7 @@
 
 from command_endpoints.ha_rest import HomeAssistantRestEndpoint
 from command_endpoints.ha_ws import HomeAssistantWebSocketEndpoint, HomeAssistantWebSocketEndpointNotSupportedException
+from command_endpoints.mqtt import MqttConfig, MqttEndpoint
 from command_endpoints.openhab import OpenhabEndpoint
 from command_endpoints.rest import RestEndpoint
 
@@ -442,6 +443,22 @@ def init_command_endpoint(app):
             except HomeAssistantWebSocketEndpointNotSupportedException:
                 app.command_endpoint = HomeAssistantRestEndpoint(host, port, tls, token)
 
+        elif user_config["command_endpoint"] == "MQTT":
+            mqtt_config = MqttConfig()
+            mqtt_config.set_auth_type(user_config["mqtt_auth_type"])
+            mqtt_config.set_hostname(user_config["mqtt_host"])
+            mqtt_config.set_port(user_config["mqtt_port"])
+            mqtt_config.set_tls(user_config["mqtt_tls"])
+            mqtt_config.set_topic(user_config["mqtt_topic"])
+
+            if 'mqtt_password' in user_config:
+                mqtt_config.set_password(user_config['mqtt_password'])
+
+            if 'mqtt_username' in user_config:
+                mqtt_config.set_username(user_config['mqtt_username'])
+
+            app.command_endpoint = MqttEndpoint(mqtt_config)
+
         elif user_config["command_endpoint"] == "openHAB":
             app.command_endpoint = OpenhabEndpoint(user_config["openhab_url"], user_config["openhab_token"])
 

command_endpoints/mqtt.py

@@ -0,0 +1,106 @@
+import asyncio
+import json
+import logging
+import paho.mqtt.client as mqtt
+from . import CommandEndpoint, CommandEndpointConfigException, CommandEndpointResult, CommandEndpointRuntimeException
+from enum import Enum
+
+
+class MqttAuthType(Enum):
+    NONE = 1
+    USERPW = 2
+
+
+class MqttConfig:
+    auth_type: Enum = MqttAuthType.NONE
+    hostname: str = None
+    password: str = None
+    port: int = 8883
+    tls: bool = True
+    topic: str = None
+    username: str = None
+
+    log = logging.getLogger("WAS")
+
+    def set_auth_type(self, auth_type=MqttAuthType.NONE):
+        self.log.debug(f"setting auth type: {auth_type}")
+        self.auth_type = MqttAuthType[auth_type.upper()]
+
+    def set_hostname(self, hostname=None):
+        self.hostname = hostname
+
+    def set_password(self, password=None):
+        self.password = password
+
+    def set_port(self, port=8883):
+        self.port = port
+
+    def set_tls(self, tls=True):
+        self.tls = tls
+
+    def set_topic(self, topic=None):
+        self.topic = topic
+
+    def set_username(self, username=None):
+        self.username = username
+
+    def validate(self):
+        if self.auth_type == MqttAuthType.USERPW:
+            if self.password is None:
+                raise CommandEndpointConfigException("User/Password auth enabled without password")
+            if self.username is None:
+                raise CommandEndpointConfigException("User/Password auth enabled without password")
+
+
+class MqttEndpoint(CommandEndpoint):
+    name = "MQTT"
+
+    def __init__(self, config):
+        self.config = config
+        self.config.validate()
+        self.mqtt_client = None
+
+        loop = asyncio.get_event_loop()
+        self.task = loop.create_task(self.connect())
+
+    async def connect(self):
+        try:
+            self.mqtt_client = mqtt.Client()
+            self.mqtt_client.on_connect = self.cb_connect
+            self.mqtt_client.on_msg = self.cb_msg
+            if self.config.username is not None and self.config.password is not None:
+                self.mqtt_client.username_pw_set(self.config.username, self.config.password)
+            if self.config.tls:
+                self.mqtt_client.tls_set()
+            self.mqtt_client.connect_async(self.config.hostname, self.config.port, 60)
+            self.mqtt_client.loop_start()
+        except Exception as e:
+            self.log.info(f"{self.name}: exception occurred: {e}")
+            await asyncio.sleep(1)
+
+    def cb_connect(self, client, userdata, flags, rc):
+        self.log.info(f"MQTT connected")
+        client.subscribe(self.config.topic)
+
+    def cb_msg(self, client, userdata, msg):
+        self.log.info(f"cb_msg: topic={msg.topic} payload={msg.payloud}")
+
+    def parse_response(self, response):
+        res = CommandEndpointResult()
+        if response.ok:
+            res.ok = True
+            if len(res.speech) > 0:
+                res.speech = response.text
+            else:
+                res.speech = "Success!"
+
+        return json.dumps({"result": res.__dict__})
+
+    def send(self, data=None, jsondata=None, ws=None):
+        try:
+            if jsondata is not None:
+                self.mqtt_client.publish(self.config.topic, payload=json.dumps(jsondata))
+            else:
+                self.mqtt_client.publish(self.config.topic, payload=data)
+        except Exception as e:
+            raise CommandEndpointRuntimeException(e)

requirements.txt

@@ -17,6 +17,7 @@ Jinja2==3.1.2
 MarkupSafe==2.1.3
 num2words==0.5.12
 orjson==3.9.7
+paho-mqtt==1.6.1
 pydantic==2.3.0
 pydantic-extra-types==2.1.0
 pydantic-settings==2.0.3