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

Pause added to stop changing log window

Browse Source
This commit is contained in:
Dirk Alders 2025-07-28 20:46:32 +02:00
parent 6bc38b0130
commit fa7c63f74b
1 changed files with 11 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
mqtt_sniffer.py
View File

@ -55,9 +55,10 @@ class MqttSniffer(App):
CSS_PATH = "style.tcss" CSS_PATH = "style.tcss"
BINDINGS = [ BINDINGS = [
("q", "quit", "Quit"), ("q", "quit", "Quit"),
("c", "clear_screen", "Clear") ("c", "clear_screen", "Clear"),
("p", "pause", "Pause")
] ]
MAX_LOGS = 1000 MAX_LOGS = 5000
def __init__(self, args, password): def __init__(self, args, password):
super().__init__() super().__init__()
@ -68,6 +69,7 @@ class MqttSniffer(App):
# #
self.mqtt = None self.mqtt = None
self.all_logs = [] self.all_logs = []
self.__logging_enabled__ = True
self.__topic_selection__ = MultiSelect((), prompt="Full", id="topic_filter") self.__topic_selection__ = MultiSelect((), prompt="Full", id="topic_filter")
self.send_topic = "" self.send_topic = ""
self.send_payload = "" self.send_payload = ""
@ -113,12 +115,18 @@ class MqttSniffer(App):
f"{repr(record.payload)}" f"{repr(record.payload)}"
) )
self.log_display.write(message) if self.__logging_enabled__:
self.log_display.write(message)
def action_clear_screen(self): def action_clear_screen(self):
self.all_logs = [] self.all_logs = []
self._update_display() self._update_display()
def action_pause(self):
self.__logging_enabled__ = not self.__logging_enabled__
if self.__logging_enabled__:
self._update_display()
def _update_display(self): def _update_display(self):
"""Clean the display and render all mqtt messages based on the current filters.""" """Clean the display and render all mqtt messages based on the current filters."""
self.log_display.clear() self.log_display.clear()