default value handling added and pep8

This commit is contained in:
Dirk Alders 2023-08-01 08:12:06 +02:00
parent c9103de08e
commit 7c00cf30f0
4 changed files with 48 additions and 7 deletions

17
.vscode/launch.json vendored Normal file
View File

@ -0,0 +1,17 @@
{
// Verwendet IntelliSense zum Ermitteln möglicher Attribute.
// Zeigen Sie auf vorhandene Attribute, um die zugehörigen Beschreibungen anzuzeigen.
// Weitere Informationen finden Sie unter https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Python: Main File execution",
"type": "python",
"request": "launch",
"program": "${workspaceFolder}/mqtt_sniffer.py",
"console": "externalTerminal",
"args": ["-n", "-u", "smarthome", "-t", "es.*t", "-l"],
"justMyCode": true
}
]
}

15
.vscode/settings.json vendored Normal file
View File

@ -0,0 +1,15 @@
{
"python.defaultInterpreterPath": "./venv/bin/python",
"autopep8.args": ["--max-line-length=150"],
"python.formatting.provider": "none",
"[python]": {
"editor.defaultFormatter": "ms-python.python",
"editor.formatOnSave": true
},
"editor.formatOnSave": true,
"editor.fontSize": 14,
"emmet.includeLanguages": { "django-html": "html" },
"python.testing.pytestArgs": ["-v", "--cov", "--cov-report=xml", "__test__"],
"python.testing.unittestEnabled": false,
"python.testing.pytestEnabled": true,
}

@ -1 +1 @@
Subproject commit fa26893496c744f416a1ee0193824075b3709330 Subproject commit c9d7f78f6d0feeeb5e14e4ca6a8ca22ce1c81165

View File

@ -9,17 +9,22 @@ import time
VERSION = "0.1.0" VERSION = "0.1.0"
STARTTIME = time.time() STARTTIME = time.time()
# TODO: Implement default values for bottombar_entries
HELPTEXT = """ HELPTEXT = """
F1: Get this help message F1: Get this help message
F2: Set a filter (regular expression) for the topic of a message F2: Set a filter (regular expression) for the topic of a message
Examples: Examples:
* "/gfw/.*/videv" Get everything with "/gfw/" before "/videv" * "/gfw/.*/main_light/" to get everything with "/gfw/" before "/main_light/"
F9: Start / Stop logging to mqtt_sniffer.log * "^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 F12: Quit the mqtt sniffer
'c': Clear screen
'q': Quit
""" """
def rx_mqtt(mc, userdata, message): def rx_mqtt(mc, userdata, message):
global my_bb global my_bb
global logfile global logfile
@ -46,27 +51,31 @@ def rx_mqtt(mc, userdata, message):
logfile.write("%9.04f::%s::%s\n" % (ts, message.topic, data)) logfile.write("%9.04f::%s::%s\n" % (ts, message.topic, data))
logfile.flush() logfile.flush()
if __name__ == "__main__": if __name__ == "__main__":
parser = argparse.ArgumentParser(description='This is a mqtt sniffer.') 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('-f', dest='hostname', default='localhost', help='Hostname of the mqtt server')
parser.add_argument('-p', dest='port', default=1883, help='Port of the mqtt server') parser.add_argument('-p', dest='port', default=1883, help='Port of the mqtt server')
parser.add_argument('-n', dest='no_credentials', action='store_true', help='Avoid asking for credentials') parser.add_argument('-n', dest='no_credentials', action='store_true', help='Avoid asking for credentials')
parser.add_argument('-u', dest='username', default=None, help='Set username for mqtt server') parser.add_argument('-u', dest='username', default=None, help='Set username for mqtt server')
parser.add_argument('-t', dest='topicfilter', default="", help='Set topic filter')
parser.add_argument('-l', dest='logtofile', action='store_true', help='Enable logging to file')
args = parser.parse_args() args = parser.parse_args()
if not args.no_credentials: if not args.no_credentials:
if args.username == None: if args.username == None:
args.username = input("Username: ") args.username = input("Username: ")
password = getpass.getpass(prompt='Password: ', stream=None) password = getpass.getpass(prompt='Password: ', stream=None)
else: else:
args.username = None
password = None password = None
my_bb = BottomBar(VERSION, label='MQTT-Sniffer') 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('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') 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('quit', 12, my_bb.FUNC_QUIT, "Quit", label='[F12]', right=True)
my_bb.add_entry('log2file', 9, my_bb.FUNC_BOOL, label='[F9] Log2File', right=True) my_bb.add_entry('log2file', 9, my_bb.FUNC_BOOL, label='[F9] Log2File', default=args.logtofile, right=True)
with open('mqtt_sniffer.log', 'w') as logfile: with open('mqtt_sniffer.log', 'w') as logfile:
mc = mqtt.mqtt_client("mqtt_sniffer", args.hostname, port=args.port, username=args.username, password=password) mc = mqtt.mqtt_client("mqtt_sniffer", args.hostname, port=args.port, username=args.username, password=password)
mc.add_callback("#", rx_mqtt) mc.add_callback("#", rx_mqtt)