application/mqtt_sniffer
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Cleanup

Browse Source
This commit is contained in:
Dirk Alders 2025-07-19 20:07:17 +02:00
parent 68e5dfc073
commit 745e8e9524
1 changed files with 15 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
mqtt_sniffer.py
View File

@ -7,26 +7,27 @@ import time
from textual.app import App, ComposeResult from textual.app import App, ComposeResult
from textual.containers import Vertical from textual.containers import Vertical
from textual.widgets import Footer, Header, Input, RichLog, Checkbox from textual.widgets import Footer, Header, Input, RichLog
class LogReceiver(object): # TODO: Usage of multiple regex for topic filter
"""Ein Thread, der auf eingehende Log-Nachrichten lauscht.""" # TODO: Integrate sending of message (topic + payload)
class MqttReceiver(object):
def __init__(self, app): def __init__(self, app):
self.app = app self.app = app
args = app.args args = app.args
password = app.password password = app.password
self.mqtt_client = mqtt.mqtt_client('mqttsniffer', args.hostname, args.port, username=args.username, password=password) self.mqtt_client = mqtt.mqtt_client('mqtt_sniffer', args.hostname, args.port, username=args.username, password=password)
self.mqtt_client.add_callback("#", self.__rx__) self.mqtt_client.add_callback("#", self.__rx__)
def __rx__(self, client, userdata, message): def __rx__(self, client, userdata, message):
self.app.call_from_thread(self.app.add_log, message) self.app.call_from_thread(self.app.add_log, message)
class LogViewerApp(App): class MqttSniffer(App):
"""Eine Textual-App zum Anzeigen und Filtern von Python-Logs.""" """a textual application for viewing mqtt messages."""
CSS_PATH = "style.tcss" CSS_PATH = "style.tcss"
BINDINGS = [("q", "quit", "Quit")] BINDINGS = [("q", "quit", "Quit")]
@ -42,7 +43,7 @@ class LogViewerApp(App):
self.log_display = RichLog(highlight=True, markup=True) self.log_display = RichLog(highlight=True, markup=True)
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
"""Erstellt die Widgets für die App.""" """Create the widgets for the application."""
yield Header(name="Python Log Viewer") yield Header(name="Python Log Viewer")
with Vertical(id="app-grid"): with Vertical(id="app-grid"):
yield self.log_display yield self.log_display
@ -51,11 +52,11 @@ class LogViewerApp(App):
yield Footer() yield Footer()
def on_mount(self) -> None: def on_mount(self) -> None:
"""Startet den Log-Empfänger-Thread.""" """start the mqtt receiver."""
log_receiver = LogReceiver(self) MqttReceiver(self)
def add_log(self, record) -> None: def add_log(self, record) -> None:
"""Fügt einen neuen Log-Eintrag hinzu und aktualisiert die Anzeige.""" """Add new mqt messages and update the tui."""
asctime = time.asctime() asctime = time.asctime()
self.all_logs.append((asctime, record)) self.all_logs.append((asctime, record))
if len(self.all_logs) > self.MAX_LOGS: if len(self.all_logs) > self.MAX_LOGS:
@ -64,13 +65,12 @@ class LogViewerApp(App):
def _apply_filters_to_log(self, data: logging.LogRecord): def _apply_filters_to_log(self, data: logging.LogRecord):
asctime, record = data asctime, record = data
"""Prüft einen einzelnen Log-Eintrag gegen die Filter und zeigt ihn ggf. an.""" """filter the mqtt messages."""
topic_match = False topic_match = False
for topic_filter in self.topic_filter.split(","): for topic_filter in self.topic_filter.split(","):
topic_match |= topic_filter.lower() in record.topic.lower() topic_match |= topic_filter.lower() in record.topic.lower()
if topic_match: if topic_match:
try: try:
payload = json.loads(record.payload) payload = json.loads(record.payload)
except: except:
@ -85,20 +85,18 @@ class LogViewerApp(App):
self.log_display.write(message) self.log_display.write(message)
def _update_display(self): def _update_display(self):
"""Löscht die Anzeige und rendert alle Logs basierend auf den aktuellen Filtern neu.""" """Clean the display and render all mqtt messages based on the current filters."""
self.log_display.clear() self.log_display.clear()
for record in self.all_logs: for record in self.all_logs:
self._apply_filters_to_log(record) self._apply_filters_to_log(record)
def on_input_changed(self, message: Input.Changed) -> None: def on_input_changed(self, message: Input.Changed) -> None:
"""Aktualisiert die Filter und die Anzeige, wenn der Benutzer tippt.""" """Update the filter and filtered messages after the user did changes."""
if message.input.id == "topic_filter": if message.input.id == "topic_filter":
self.topic_filter = message.value self.topic_filter = message.value
self._update_display() self._update_display()
self._update_display()
if __name__ == "__main__": if __name__ == "__main__":
# #
@ -128,5 +126,5 @@ if __name__ == "__main__":
# #
# Start Application # Start Application
# #
app = LogViewerApp(args, password) app = MqttSniffer(args, password)
app.run() app.run()