MQTT Connection

[MRIIOT]

Example of building a treeview from MQTT topics.

import os
import sys
import uuid
import json
import time
import aiomqtt
from textual import work
from textual.app import App
from textual.app import ComposeResult
from textual.messages import Message
from textual.widget import Widget
from textual.widgets import Log, Welcome, Label, Tree
from textual.worker import get_current_worker


class ExampleApp(App):

    class MQTTMessage(Message):
        def __init__(self, topic: str, payload: dict):
            super().__init__()
            self.topic = topic
            self.payload = payload

    def on_mount(self):
        self.start_worker()

    def compose(self) -> ComposeResult:
        tree: Tree[dict] = Tree("Broker")
        tree.root.expand()
        self.devices_online = tree.root.add("Online", expand=True)
        self.devices_offline = tree.root.add("Offline", expand=True)
        yield tree

    def on_example_app_mqttmessage(self, message: MQTTMessage) -> None:
        device_id = message.topic.split("/", 2)[1]
        is_online = message.payload["v"] is True
        belongs_to = self.devices_online if is_online else self.devices_offline
        color = "green" if is_online else "red"
        found_node_online = [node for node in self.devices_online.children if node.data["serial"] == device_id]
        found_node_offline = [node for node in self.devices_offline.children if node.data["serial"] == device_id]
        found_node = found_node_online or found_node_offline

        if not found_node:
            belongs_to.add_leaf(f"[{color}]{device_id}", data={"serial": device_id})
        elif is_online and found_node_offline:
            found_node_offline[0].remove()
            belongs_to.add_leaf(f"[{color}]{device_id}", data={"serial": device_id})
        elif not is_online and found_node_online:
            found_node_online[0].remove()
            belongs_to.add_leaf(f"[{color}]{device_id}", data={"serial": device_id})

    @work(thread=False)
    async def start_worker(self) -> None:
        async with aiomqtt.Client(
                hostname="mybroker",
                port=1883,
                username=None,
                password=None,
                identifier=str(uuid.uuid4),
                clean_session=True,
                keepalive=60) as client:
            await client.subscribe(topic="sharc/+/evt/avail", qos=0)

            async for message in client.messages:
                self.post_message(self.MQTTMessage(message.topic.value, json.loads(message.payload.decode("utf-8"))))


if __name__ == "__main__":
    # https://github.com/sbtinstruments/aiomqtt#note-for-windows-users
    # Change to the "Selector" event loop if platform is Windows
    if sys.platform.lower() == "win32" or os.name.lower() == "nt":
        from asyncio import set_event_loop_policy, WindowsSelectorEventLoopPolicy
        set_event_loop_policy(WindowsSelectorEventLoopPolicy())
    app = ExampleApp()
    app.run()