From 68e5dfc0737bc88dcda633d3e218424a0dd00ce2 Mon Sep 17 00:00:00 2001 From: Dirk Alders Date: Sat, 19 Jul 2025 19:44:14 +0200 Subject: [PATCH] Updated to textual --- .gitmodules | 3 - console_bottombar | 1 - init_venv | 8 --- mqtt_sniffer.py | 167 ++++++++++++++++++++++++++-------------------- reposinit | 6 ++ requirements.txt | 2 + style.tcss | 17 +++++ 7 files changed, 121 insertions(+), 83 deletions(-) delete mode 160000 console_bottombar delete mode 100755 init_venv create mode 100755 reposinit create mode 100644 requirements.txt create mode 100644 style.tcss diff --git a/.gitmodules b/.gitmodules index 2a59480..6d8647c 100644 --- a/.gitmodules +++ b/.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 diff --git a/console_bottombar b/console_bottombar deleted file mode 160000 index c9d7f78..0000000 --- a/console_bottombar +++ /dev/null @@ -1 +0,0 @@ -Subproject commit c9d7f78f6d0feeeb5e14e4ca6a8ca22ce1c81165 diff --git a/init_venv b/init_venv deleted file mode 100755 index 428ff04..0000000 --- a/init_venv +++ /dev/null @@ -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 diff --git a/mqtt_sniffer.py b/mqtt_sniffer.py index 25c367e..9ec2655 100644 --- a/mqtt_sniffer.py +++ b/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() diff --git a/reposinit b/reposinit new file mode 100755 index 0000000..169394c --- /dev/null +++ b/reposinit @@ -0,0 +1,6 @@ +#!/bin/sh +# +git submodule init +git submodule update +echo "* Creating virtual env" +mkvenv diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..c2fb811 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +textual + diff --git a/style.tcss b/style.tcss new file mode 100644 index 0000000..159b353 --- /dev/null +++ b/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%; +}