115 lines
4.1 KiB
Python
115 lines
4.1 KiB
Python
import config
|
|
import mqtt
|
|
import readline
|
|
import sys
|
|
import report
|
|
import logging
|
|
import devices
|
|
import json
|
|
|
|
|
|
if __name__ == "__main__":
|
|
report.stdoutLoggingConfigure([[config.APP_NAME, logging.INFO], ], fmt=report.SHORT_FMT)
|
|
mc = mqtt.mqtt_client(
|
|
host=config.MQTT_SERVER,
|
|
port=config.MQTT_PORT,
|
|
username=config.MQTT_USER,
|
|
password=config.MQTT_PASSWORD,
|
|
name=config.APP_NAME + '_devicetest'
|
|
)
|
|
#
|
|
devicedict = {}
|
|
for device in [
|
|
# devices.shelly_pro3(mc, "shellies/gfw/pro3"),
|
|
# devices.brennenstuhl_heatingvalve(mc, "zigbee_raspi/heatvlv"),
|
|
# devices.silvercrest_button(mc, "zigbee_raspi/button"),
|
|
devices.hue_sw_br_ct(mc, "zigbee_ffe/kitchen/main_light_1"),
|
|
]:
|
|
devicedict[device.topic.replace("/", "_")] = device
|
|
#
|
|
COMMANDS = ['quit', 'help', 'action']
|
|
for key in devicedict:
|
|
device = devicedict[key]
|
|
for cmd in device.__class__.__dict__:
|
|
obj = getattr(device, cmd)
|
|
if callable(obj) and not cmd.startswith("_"):
|
|
COMMANDS.append(key + "." + cmd)
|
|
#
|
|
|
|
def reduced_list(text):
|
|
"""
|
|
Create reduced completation list
|
|
"""
|
|
reduced_list = {}
|
|
for cmd in COMMANDS:
|
|
next_dot = cmd[len(text):].find('.')
|
|
if next_dot >= 0:
|
|
reduced_list[cmd[:len(text) + next_dot + 1]] = None
|
|
else:
|
|
reduced_list[cmd] = None
|
|
return reduced_list.keys()
|
|
|
|
def completer(text, state):
|
|
"""
|
|
Our custom completer function
|
|
"""
|
|
options = [x for x in reduced_list(text) if x.startswith(text)]
|
|
return options[state]
|
|
|
|
def complete(text, state):
|
|
for cmd in COMMANDS:
|
|
if cmd.startswith(text):
|
|
if not state:
|
|
hit = ""
|
|
index = 0
|
|
sub_list = cmd.split('.')
|
|
while len(text) >= len(hit):
|
|
hit += sub_list[index] + '.'
|
|
return hit # cmd
|
|
else:
|
|
state -= 1
|
|
|
|
if len(sys.argv) == 1:
|
|
readline.parse_and_bind("tab: complete")
|
|
readline.set_completer(completer)
|
|
print("\nEnter command: ")
|
|
while True:
|
|
userfeedback = input('')
|
|
command = userfeedback.split(' ')[0]
|
|
if userfeedback == 'quit':
|
|
break
|
|
elif userfeedback == 'help':
|
|
print("Help is not yet implemented!")
|
|
else:
|
|
try:
|
|
key, command = userfeedback.split(".", 1)
|
|
except ValueError:
|
|
print("Unknown device.")
|
|
else:
|
|
device = devicedict[key]
|
|
try:
|
|
command, params = command.split(" ", 1)
|
|
except ValueError:
|
|
params = None
|
|
try:
|
|
obj = getattr(device, command)
|
|
except AttributeError:
|
|
print("Unknown command.")
|
|
else:
|
|
if params is not None:
|
|
params = params.replace("True", "true")
|
|
params = params.replace("False", "false")
|
|
params = params.replace("None", "null")
|
|
params = params.replace(",", " ")
|
|
params = params.split(" ")
|
|
params = " ".join([p for p in params if p])
|
|
try:
|
|
params = json.loads("[" + params.replace(" ", ", ") + "]")
|
|
except json.decoder.JSONDecodeError:
|
|
print("You need to give python like parameters (e.g. 'test' for a string containing test).")
|
|
params = None
|
|
try:
|
|
obj(*params)
|
|
except TypeError:
|
|
print("Give the correct parameters to execute.")
|