68 lines
1.8 KiB
Python
68 lines
1.8 KiB
Python
import argparse
|
|
from console_bottombar import BottomBar
|
|
import os
|
|
import re
|
|
import sys
|
|
import tail
|
|
import task
|
|
|
|
VERSION = "0.1.0"
|
|
|
|
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"
|
|
F12: Quit ftail
|
|
|
|
'c': Clear screen
|
|
'q': Quit
|
|
"""
|
|
|
|
|
|
def tail_parser(line):
|
|
global my_bb
|
|
|
|
try:
|
|
match = len(re.findall(my_bb.get_entry('msg_re'), line)) > 0
|
|
except re.error:
|
|
print('No valid regular expression (%s). No filter active.' % my_bb.get_entry('msg_re'))
|
|
match = True
|
|
|
|
if match:
|
|
sys.stdout.write(line)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description='This is a mqtt sniffer.')
|
|
parser.add_argument('logfile') # , nargs='+' for >= 1
|
|
|
|
args = parser.parse_args()
|
|
|
|
|
|
if not os.access(args.logfile, os.R_OK):
|
|
print("Error: File %s is not readable!" % logfile)
|
|
sys.exit(1)
|
|
else:
|
|
|
|
#
|
|
# Background tail task
|
|
#
|
|
tp = tail.Tail(args.logfile)
|
|
tp.register_callback(tail_parser)
|
|
my_tail = task.delayed(0.1, tp.follow, 0.1)
|
|
|
|
#
|
|
# Bottombar
|
|
#
|
|
my_bb = BottomBar(VERSION, label='ftail')
|
|
my_bb.add_entry('help', 1, my_bb.FUNC_INFO, '[F1] Help', infotext=HELPTEXT)
|
|
my_bb.add_entry('msg_re', 2, my_bb.FUNC_TEXT, label='[F2] Filter', default="")
|
|
my_bb.add_entry('quit', 12, my_bb.FUNC_QUIT, "Quit", label='[F12]', right=True)
|
|
|
|
my_tail.run()
|
|
my_bb.run()
|
|
|
|
my_tail.stop() |