Updated to textual

.gitmodules

@@ -1,6 +1,3 @@
 [submodule "mqtt"]
 	path = mqtt
 	url = https://git.mount-mockery.de/pylib/mqtt.git
-[submodule "console_bottombar"]
-	path = console_bottombar
-	url = https://git.mount-mockery.de/pylib/console_bottombar.git

console_bottombar

@@ -1 +0,0 @@
-Subproject commit c9d7f78f6d0feeeb5e14e4ca6a8ca22ce1c81165

init_venv

@@ -1,8 +0,0 @@
-#!/bin/sh
-#
-BASEPATH=`realpath $(dirname $0)`
-
-python3 -m venv $BASEPATH/venv
-$BASEPATH/venv/bin/pip install --upgrade pip
-find $BASEPATH -name requirements.txt | xargs -L 1 $BASEPATH/venv/bin/pip install -r
-$BASEPATH/venv/bin/pip list --outdated --format=json | jq -r '.[] | .name'|xargs -n1 $BASEPATH/venv/bin/pip install -U

mqtt_sniffer.py

@@ -1,78 +1,109 @@
 import argparse
-try:
-    import bottombar as bb
-except ModuleNotFoundError:
-    bb = None
-try:
-    from console_bottombar import BottomBar
-except ModuleNotFoundError:
-    BottomBar = None
 import getpass
+import logging
 import json
 import mqtt
-import os
-import re
 import time
 
-VERSION = "0.2.0"
-logfile = None
-
-HELPTEXT = """
-F1: Get this help message
-F2: Set a filter (regular expression) for the topic of a message
-    Examples:
-    * "/gfw/.*/main_light/" to get everything with "/gfw/" before "/main_light/"
-    * "^zigbee.*(?>!logging)$" to get everything starting with "zigbee" and not ending with "logging"
-    * "^(?!shellies).*/dirk/.*temperature$" to get everything not starting with "shellies" followed by "/dirk/" and ends with "temperature"
-F9: Start / Stop logging to mqtt-sniffer.log
-F12: Quit the mqtt sniffer
-
-'c': Clear screen
-'q': Quit
-"""
+from textual.app import App, ComposeResult
+from textual.containers import Vertical
+from textual.widgets import Footer, Header, Input, RichLog, Checkbox
 
 
-def msg_print_log(message, topic_regex, log2file):
-    global logfile
+class LogReceiver(object):
+    """Ein Thread, der auf eingehende Log-Nachrichten lauscht."""
 
-    try:
-        match = len(re.findall(topic_regex, message.topic)) > 0
-    except re.error:
-        print('No valid regular expression (%s). No filter active.' % topic_regex)
-        match = True
+    def __init__(self, app):
+        self.app = app
+        args = app.args
+        password = app.password
 
-    data = None
-    try:
-        data = json.loads(message.payload)
-    except:
-        pass
-    try:
-        data = message.payload.decode("utf-8")
-    except:
-        data = message.payload
+        self.mqtt_client = mqtt.mqtt_client('mqttsniffer', args.hostname, args.port, username=args.username, password=password)
+        self.mqtt_client.add_callback("#", self.__rx__)
 
-    if match:
-        print("%25s::%75s::%s" % (time.asctime(), message.topic, data))
-        if log2file:
-            if logfile is None:
-                logfile = open('mqtt_sniffer.log', 'w')
-            logfile.write("%25s::%s::%s\n" % (time.asctime(), message.topic, data))
-            logfile.flush()
+    def __rx__(self, client, userdata, message):
+        self.app.call_from_thread(self.app.add_log, message)
 
 
+class LogViewerApp(App):
+    """Eine Textual-App zum Anzeigen und Filtern von Python-Logs."""
 
-def rx_mqtt(mc, userdata, message):
-    global my_bb
-    global args
+    CSS_PATH = "style.tcss"
+    BINDINGS = [("q", "quit", "Quit")]
+    MAX_LOGS = 50
 
-    if BottomBar is not None and bb is not None:
-        msg_print_log(message, my_bb.get_entry('msg_re'), my_bb.get_entry('log2file'))
-        bb.redraw()
-    else:
-        msg_print_log(message, args.topicfilter, args.logtofile)
+    def __init__(self, args, password):
+        super().__init__()
+        self.args = args
+        self.password = password
+        #
+        self.all_logs = []
+        self.topic_filter = ""
+        self.log_display = RichLog(highlight=True, markup=True)
+
+    def compose(self) -> ComposeResult:
+        """Erstellt die Widgets für die App."""
+        yield Header(name="Python Log Viewer")
+        with Vertical(id="app-grid"):
+            yield self.log_display
+        with Vertical(id="bottom-bar"):
+            yield Input(placeholder="topic filter...", id="topic_filter")
+        yield Footer()
+
+    def on_mount(self) -> None:
+        """Startet den Log-Empfänger-Thread."""
+        log_receiver = LogReceiver(self)
+
+    def add_log(self, record) -> None:
+        """Fügt einen neuen Log-Eintrag hinzu und aktualisiert die Anzeige."""
+        asctime = time.asctime()
+        self.all_logs.append((asctime, record))
+        if len(self.all_logs) > self.MAX_LOGS:
+            self.all_logs = self.all_logs[-self.MAX_LOGS:]
+        self._apply_filters_to_log((asctime, record))
+
+    def _apply_filters_to_log(self, data: logging.LogRecord):
+        asctime, record = data
+        """Prüft einen einzelnen Log-Eintrag gegen die Filter und zeigt ihn ggf. an."""
+        topic_match = False
+        for topic_filter in self.topic_filter.split(","):
+            topic_match |= topic_filter.lower() in record.topic.lower()
+
+        if topic_match:
+
+            try:
+                payload = json.loads(record.payload)
+            except:
+                payload = record.payload.decode('utf-8')
+
+            message = (
+                f"[[dim]{asctime}[/dim]] "
+                f"[bold]{record.topic}[/bold] - "
+                f"{repr(payload)}"
+            )
+
+            self.log_display.write(message)
+
+    def _update_display(self):
+        """Löscht die Anzeige und rendert alle Logs basierend auf den aktuellen Filtern neu."""
+        self.log_display.clear()
+        for record in self.all_logs:
+            self._apply_filters_to_log(record)
+
+    def on_input_changed(self, message: Input.Changed) -> None:
+        """Aktualisiert die Filter und die Anzeige, wenn der Benutzer tippt."""
+        if message.input.id == "topic_filter":
+            self.topic_filter = message.value
+        
+        self._update_display()
+
+        self._update_display()
 
 
 if __name__ == "__main__":
+    #
+    # Parse Arguments
+    #
     parser = argparse.ArgumentParser(description='This is a mqtt sniffer.')
     parser.add_argument('-f', dest='hostname', default='localhost', help='Hostname of the mqtt server')
     parser.add_argument('-p', dest='port', default=1883, type=int, help='Port of the mqtt server')
@@ -83,6 +114,9 @@ if __name__ == "__main__":
 
     args = parser.parse_args()
 
+    #
+    # Ask for credentials
+    #
     if not args.no_credentials:
         if args.username == None:
             args.username = input("Username: ")
@@ -91,17 +125,8 @@ if __name__ == "__main__":
         args.username = None
         password = None
 
-    if BottomBar != None:
-        my_bb = BottomBar(VERSION, label='MQTT-Sniffer')
-        my_bb.add_entry('help', 1, my_bb.FUNC_INFO, label='[F1] Help', infotext=HELPTEXT)
-        my_bb.add_entry('msg_re', 2, my_bb.FUNC_TEXT, label='[F2] Filter', default=args.topicfilter)
-        my_bb.add_entry('quit', 12, my_bb.FUNC_QUIT, "Quit", label='[F12]', right=True)
-        my_bb.add_entry('log2file', 9, my_bb.FUNC_BOOL, label='[F9] Log2File', default=args.logtofile, right=True)
-
-    mc = mqtt.mqtt_client("mqtt_sniffer", args.hostname, port=args.port, username=args.username, password=password)
-    mc.add_callback("#", rx_mqtt)
-    if BottomBar != None:
-        my_bb.run()
-    else:
-        while True:
-            time.sleep(1)
+    #
+    # Start Application
+    #
+    app = LogViewerApp(args, password)
+    app.run()

reposinit

@@ -0,0 +1,6 @@
+#!/bin/sh
+#
+git submodule init
+git submodule update
+echo "* Creating virtual env"
+mkvenv

requirements.txt

@@ -0,0 +1,2 @@
+textual
+

style.tcss

@@ -0,0 +1,17 @@
+#app-grid {
+    layout: vertical;
+    overflow-y: scroll;
+    scrollbar-gutter: stable;
+    height: 89%;
+}
+
+#bottom-bar {
+    layout: grid;
+    grid-size: 1;
+    height: 11%;
+    border-top: solid $primary;
+}
+
+#module_filter {
+    width: 100%;
+}