test and simulation extracted to own project space
This commit is contained in:
parent
6c1f81f0be
commit
da9deef32e
@ -1,832 +0,0 @@
|
||||
import colored
|
||||
import copy
|
||||
import json
|
||||
import task
|
||||
import time
|
||||
|
||||
COLOR_GUI_ACTIVE = colored.fg("light_blue")
|
||||
COLOR_GUI_PASSIVE = COLOR_GUI_ACTIVE + colored.attr("dim")
|
||||
COLOR_SHELLY = colored.fg("light_magenta")
|
||||
COLOR_POWERPLUG = colored.fg("light_cyan")
|
||||
COLOR_LIGHT_ACTIVE = colored.fg("yellow")
|
||||
COLOR_LIGHT_PASSIVE = COLOR_LIGHT_ACTIVE + colored.attr("dim")
|
||||
COLOR_MOTION_SENSOR = colored.fg("dark_orange_3b")
|
||||
COLOR_HEATING_VALVE = colored.fg("red")
|
||||
COLOR_REMOTE = colored.fg("green")
|
||||
|
||||
|
||||
def payload_filter(payload):
|
||||
try:
|
||||
return json.loads(payload)
|
||||
except json.decoder.JSONDecodeError:
|
||||
return payload.decode("utf-8")
|
||||
|
||||
|
||||
def command_int_value(value):
|
||||
try:
|
||||
return int(value)
|
||||
except TypeError:
|
||||
print("You need to give a integer parameter not '%s'" % str(value))
|
||||
|
||||
|
||||
def command_float_value(value):
|
||||
try:
|
||||
return float(value)
|
||||
except TypeError:
|
||||
print("You need to give a integer parameter not '%s'" % str(value))
|
||||
|
||||
|
||||
def devicename(topic):
|
||||
return " - ".join(topic.split('/')[1:])
|
||||
|
||||
|
||||
def percent_bar(value):
|
||||
rv = ""
|
||||
for i in range(0, 10):
|
||||
rv += u"\u25ac" if (value - 5) > 10*i else u"\u25ad"
|
||||
return rv
|
||||
|
||||
|
||||
def print_light(color, state, topic, description, led=False):
|
||||
if led is True:
|
||||
if state is True:
|
||||
icon = colored.fg('green') + "\u2b24" + color
|
||||
else:
|
||||
icon = colored.fg('light_gray') + "\u2b24" + color
|
||||
else:
|
||||
icon = u'\u2b24' if state is True else u'\u25ef'
|
||||
print(color + 10 * ' ' + icon + 9 * ' ' + devicename(topic), description + colored.attr("reset"))
|
||||
|
||||
|
||||
def print_switch(color, state, topic, description):
|
||||
icon = u'\u25a0' if state is True else u'\u25a1'
|
||||
print(color + 10 * ' ' + icon + 9 * ' ' + devicename(topic), description + colored.attr("reset"))
|
||||
|
||||
|
||||
def print_percent(color, prefix, perc_value, value_str, topic, description):
|
||||
if len(prefix) > 1 or len(value_str) > 7:
|
||||
raise ValueError("Length of prefix (%d) > 1 or length of value_str (%d) > 7" % (len(prefix), len(value_str)))
|
||||
print(color + prefix + percent_bar(perc_value), value_str + (8 - len(value_str)) * ' ' + devicename(topic), description + colored.attr("reset"))
|
||||
|
||||
|
||||
class base(object):
|
||||
AUTOSEND = True
|
||||
COMMANDS = []
|
||||
|
||||
def __init__(self, mqtt_client, topic):
|
||||
self.mqtt_client = mqtt_client
|
||||
self.topic = topic
|
||||
#
|
||||
self.data = {}
|
||||
self.callbacks = {}
|
||||
self.names = {}
|
||||
self.commands = self.COMMANDS[:]
|
||||
#
|
||||
self.mqtt_client.add_callback(self.topic, self.__rx__)
|
||||
self.mqtt_client.add_callback(self.topic + '/#', self.__rx__)
|
||||
|
||||
def add_callback(self, key, callback, value):
|
||||
if self.callbacks.get(key) is None:
|
||||
self.callbacks[key] = []
|
||||
self.callbacks[key].append((callback, value))
|
||||
|
||||
def add_channel_name(self, key, name):
|
||||
self.names[key] = name
|
||||
|
||||
def capabilities(self):
|
||||
return self.commands
|
||||
|
||||
def store_data(self, *args, **kwargs):
|
||||
keys_changed = []
|
||||
for key in kwargs:
|
||||
if kwargs[key] is not None and kwargs[key] != self.data.get(key):
|
||||
keys_changed.append(key)
|
||||
self.data[key] = kwargs[key]
|
||||
for callback, value in self.callbacks.get(key, [(None, None)]):
|
||||
if callback is not None and (value is None or value == kwargs[key]):
|
||||
callback(self, key, kwargs[key])
|
||||
if self.AUTOSEND and len(keys_changed) > 0:
|
||||
self.__tx__(keys_changed)
|
||||
|
||||
def get_data(self, key, default=None):
|
||||
rv = self.data.get(key, default)
|
||||
try:
|
||||
rv = True if rv.lower() == 'on' else rv
|
||||
rv = False if rv.lower() == 'off' else rv
|
||||
except AttributeError:
|
||||
pass
|
||||
return rv
|
||||
|
||||
def __tx__(self, keys_changed):
|
||||
self.mqtt_client.send(self.topic, json.dumps(self.data))
|
||||
|
||||
def __rx__(self, client, userdata, message):
|
||||
print("%s: __rx__ not handled!" % self.__class__.__name__)
|
||||
|
||||
|
||||
class shelly(base):
|
||||
KEY_OUTPUT_0 = "relay/0"
|
||||
KEY_OUTPUT_1 = "relay/1"
|
||||
KEY_INPUT_0 = "input/0"
|
||||
KEY_INPUT_1 = "input/1"
|
||||
KEY_LONGPUSH_0 = "longpush/0"
|
||||
KEY_LONGPUSH_1 = "longpush/1"
|
||||
#
|
||||
INPUT_FUNC_OUT1_FOLLOW = "out1_follow"
|
||||
INPUT_FUNC_OUT1_TRIGGER = "out1_trigger"
|
||||
INPUT_FUNC_OUT2_FOLLOW = "out2_follow"
|
||||
INPUT_FUNC_OUT2_TRIGGER = "out2_trigger"
|
||||
#
|
||||
COMMANDS = [
|
||||
"get_relay_0", "toggle_relay_0",
|
||||
"get_relay_1", "toggle_relay_1",
|
||||
"get_input_0", "toggle_input_0",
|
||||
"get_input_1", "toggle_input_1",
|
||||
"trigger_long_input_0", "trigger_long_input_1",
|
||||
]
|
||||
|
||||
def __init__(self, mqtt_client, topic, input_0_func=None, input_1_func=None, output_0_auto_off=None):
|
||||
super().__init__(mqtt_client, topic)
|
||||
#
|
||||
self.store_data(**{self.KEY_OUTPUT_0: False, self.KEY_OUTPUT_1: False, self.KEY_INPUT_0: False, self.KEY_INPUT_1: False})
|
||||
self.__input_0_func = input_0_func
|
||||
self.__input_1_func = input_1_func
|
||||
self.__output_0_auto_off__ = output_0_auto_off
|
||||
if self.__output_0_auto_off__ is not None:
|
||||
self.__delayed_off__ = task.delayed(float(self.__output_0_auto_off__), self.__auto_off__, self.KEY_OUTPUT_0)
|
||||
#
|
||||
self.add_callback(self.KEY_OUTPUT_0, self.print_formatted, None)
|
||||
self.add_callback(self.KEY_OUTPUT_0, self.__start_auto_off__, True)
|
||||
self.add_callback(self.KEY_OUTPUT_0, self.__stop_auto_off__, True)
|
||||
self.add_callback(self.KEY_OUTPUT_1, self.print_formatted, None)
|
||||
#
|
||||
self.add_callback(self.KEY_INPUT_0, self.__input_function__, None)
|
||||
self.add_callback(self.KEY_INPUT_1, self.__input_function__, None)
|
||||
|
||||
def __rx__(self, client, userdata, message):
|
||||
value = payload_filter(message.payload)
|
||||
if message.topic.startswith(self.topic) and message.topic.endswith("/command"):
|
||||
key = '/'.join(message.topic.split('/')[-3:-1])
|
||||
if value == 'toggle':
|
||||
self.__toggle_data__(key)
|
||||
else:
|
||||
self.__set_data__(key, value)
|
||||
|
||||
def __tx__(self, keys_changed):
|
||||
for key in keys_changed:
|
||||
self.mqtt_client.send(self.topic + '/' + key, "on" if self.data.get(key) else "off")
|
||||
|
||||
def __input_function__(self, device, key, data):
|
||||
if key == self.KEY_INPUT_0:
|
||||
func = self.__input_0_func
|
||||
elif key == self.KEY_INPUT_1:
|
||||
func = self.__input_1_func
|
||||
else:
|
||||
func = None
|
||||
if func == self.INPUT_FUNC_OUT1_FOLLOW:
|
||||
self.__set_data__(self.KEY_OUTPUT_0, data)
|
||||
elif func == self.INPUT_FUNC_OUT1_TRIGGER:
|
||||
self.__toggle_data__(self.KEY_OUTPUT_0)
|
||||
elif func == self.INPUT_FUNC_OUT2_FOLLOW:
|
||||
self.__set_data__(self.KEY_OUTPUT_1, data)
|
||||
elif func == self.INPUT_FUNC_OUT2_TRIGGER:
|
||||
self.__toggle_data__(self.KEY_OUTPUT_1)
|
||||
|
||||
def __start_auto_off__(self, device, key, data):
|
||||
self.__stop_auto_off__(device, key, data)
|
||||
if self.__output_0_auto_off__ is not None:
|
||||
self.__delayed_off__.run()
|
||||
|
||||
def __stop_auto_off__(self, device, key, data):
|
||||
if self.__output_0_auto_off__ is not None:
|
||||
if not self.__delayed_off__._stopped:
|
||||
self.__delayed_off__.stop()
|
||||
|
||||
def __auto_off__(self, key):
|
||||
if key == self.KEY_OUTPUT_0:
|
||||
self.__set_data__(key, 'off')
|
||||
|
||||
def __set_data__(self, key, value):
|
||||
self.store_data(**{key: value == "on"})
|
||||
|
||||
def __toggle_data__(self, key):
|
||||
if key in self.data:
|
||||
self.__set_data__(key, "on" if not self.data.get(key) else "off")
|
||||
|
||||
def command(self, command):
|
||||
if command in self.COMMANDS:
|
||||
if command == self.COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_OUTPUT_0, self.data.get(self.KEY_OUTPUT_0))
|
||||
elif command == self.COMMANDS[1]:
|
||||
self.__toggle_data__(self.KEY_OUTPUT_0)
|
||||
elif command == self.COMMANDS[2]:
|
||||
self.print_formatted(self, self.KEY_OUTPUT_1, self.data.get(self.KEY_OUTPUT_1))
|
||||
elif command == self.COMMANDS[3]:
|
||||
self.__toggle_data__(self.KEY_OUTPUT_1)
|
||||
elif command == self.COMMANDS[4]:
|
||||
self.print_formatted(self, self.KEY_INPUT_0, self.data.get(self.KEY_INPUT_0))
|
||||
elif command == self.COMMANDS[5]:
|
||||
self.__toggle_data__(self.KEY_INPUT_0)
|
||||
elif command == self.COMMANDS[6]:
|
||||
self.print_formatted(self, self.KEY_INPUT_1, self.data.get(self.KEY_INPUT_1))
|
||||
elif command == self.COMMANDS[7]:
|
||||
self.__toggle_data__(self.KEY_INPUT_1)
|
||||
elif command == self.COMMANDS[8]:
|
||||
self.__toggle_data__(self.KEY_INPUT_0)
|
||||
time.sleep(0.4)
|
||||
self.__set_data__(self.KEY_LONGPUSH_0, True)
|
||||
time.sleep(0.1)
|
||||
self.__set_data__(self.KEY_LONGPUSH_0, False)
|
||||
elif command == self.COMMANDS[9]:
|
||||
self.__toggle_data__(self.KEY_INPUT_1)
|
||||
time.sleep(0.4)
|
||||
self.__set_data__(self.KEY_LONGPUSH_1, True)
|
||||
time.sleep(0.1)
|
||||
self.__set_data__(self.KEY_LONGPUSH_1, False)
|
||||
else:
|
||||
print("%s: not yet implemented!" % command)
|
||||
else:
|
||||
print("Unknown command!")
|
||||
|
||||
def print_formatted(self, device, key, value):
|
||||
if value is not None:
|
||||
info = (" - %ds" % self.__output_0_auto_off__) if self.__output_0_auto_off__ is not None and value else ""
|
||||
channel = "(%s%s)" % (self.names.get(key, key), info)
|
||||
print_light(COLOR_SHELLY, value, self.topic, channel)
|
||||
|
||||
|
||||
class my_powerplug(base):
|
||||
KEY_OUTPUT_0 = "state"
|
||||
COMMANDS = [
|
||||
"get_output", "toggle_output",
|
||||
]
|
||||
|
||||
def __init__(self, mqtt_client, topic, channel):
|
||||
super().__init__(mqtt_client, topic + '/' + "output/%d" % (channel + 1))
|
||||
#
|
||||
self.data[self.KEY_OUTPUT_0] = False
|
||||
self.add_callback(self.KEY_OUTPUT_0, self.print_formatted, None)
|
||||
|
||||
def __rx__(self, client, userdata, message):
|
||||
if message.topic == self.topic + '/set':
|
||||
payload = payload_filter(message.payload)
|
||||
if payload == "toggle":
|
||||
payload = not self.data.get(self.KEY_OUTPUT_0)
|
||||
self.store_data(**{self.KEY_OUTPUT_0: payload})
|
||||
|
||||
def __tx__(self, keys_changed):
|
||||
for key in keys_changed:
|
||||
self.mqtt_client.send(self.topic, json.dumps(self.data.get(key)))
|
||||
|
||||
def command(self, command):
|
||||
if command in self.COMMANDS:
|
||||
if command == self.COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_OUTPUT_0, self.data.get(self.KEY_OUTPUT_0))
|
||||
elif command == self.COMMANDS[1]:
|
||||
self.store_data(**{self.KEY_OUTPUT_0: not self.data.get(self.KEY_OUTPUT_0)})
|
||||
else:
|
||||
print("%s: not yet implemented!" % command)
|
||||
else:
|
||||
print("Unknown command!")
|
||||
|
||||
def print_formatted(self, device, key, value):
|
||||
if value is not None:
|
||||
print_light(COLOR_POWERPLUG, value, self.topic, "(%s)" % self.names.get(key, "State"))
|
||||
|
||||
|
||||
class silvercrest_powerplug(base):
|
||||
KEY_OUTPUT_0 = "state"
|
||||
#
|
||||
COMMANDS = [
|
||||
"get_state", "set_state", "unset_state",
|
||||
]
|
||||
|
||||
def __init__(self, mqtt_client, topic):
|
||||
super().__init__(mqtt_client, topic)
|
||||
self.add_callback(self.KEY_OUTPUT_0, self.print_formatted, None)
|
||||
#
|
||||
self.store_data(**{self.KEY_OUTPUT_0: False})
|
||||
|
||||
def __rx__(self, client, userdata, message):
|
||||
if message.topic == self.topic + '/set':
|
||||
STATES = ["on", "off", "toggle"]
|
||||
#
|
||||
state = json.loads(message.payload).get('state').lower()
|
||||
if state in STATES:
|
||||
if state == STATES[0]:
|
||||
self.store_data(**{self.KEY_OUTPUT_0: True})
|
||||
elif state == STATES[1]:
|
||||
self.store_data(**{self.KEY_OUTPUT_0: False})
|
||||
else:
|
||||
self.store_data(**{not self.data.get(self.KEY_OUTPUT_0)})
|
||||
|
||||
def __tx__(self, keys_changed):
|
||||
for key in keys_changed:
|
||||
self.mqtt_client.send(self.topic + '/' + key, "on" if self.data.get(key) else "off")
|
||||
|
||||
def command(self, command):
|
||||
if command in self.COMMANDS:
|
||||
if command == self.COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_OUTPUT_0, self.data.get(self.KEY_OUTPUT_0))
|
||||
elif command == self.COMMANDS[1]:
|
||||
self.store_data(**{self.KEY_OUTPUT_0: True})
|
||||
elif command == self.COMMANDS[2]:
|
||||
self.store_data(**{self.KEY_OUTPUT_0: False})
|
||||
else:
|
||||
print("%s: not yet implemented!" % command)
|
||||
else:
|
||||
print("Unknown command!")
|
||||
|
||||
def print_formatted(self, device, key, value):
|
||||
if value is not None:
|
||||
print_light(COLOR_POWERPLUG, value, self.topic, "(%s)" % self.names.get(key, key))
|
||||
|
||||
|
||||
class tradfri_light(base):
|
||||
KEY_OUTPUT_0 = "state"
|
||||
KEY_BRIGHTNESS = "brightness"
|
||||
KEY_COLOR_TEMP = "color_temp"
|
||||
KEY_BRIGHTNESS_MOVE = "brightness_move"
|
||||
#
|
||||
STATE_COMMANDS = ("get_state", "toggle_state", )
|
||||
BRIGHTNESS_COMMANDS = ("get_brightness", "set_brightness",)
|
||||
COLOR_TEMP_COMMANDS = ("get_color_temp", "set_color_temp",)
|
||||
|
||||
def __init__(self, mqtt_client, topic, enable_state=True, enable_brightness=False, enable_color_temp=False, send_on_power_on=True):
|
||||
super().__init__(mqtt_client, topic)
|
||||
self.send_on_power_on = send_on_power_on
|
||||
self.add_callback(self.KEY_OUTPUT_0, self.print_formatted, None)
|
||||
self.add_callback(self.KEY_BRIGHTNESS, self.print_formatted, None)
|
||||
self.add_callback(self.KEY_COLOR_TEMP, self.print_formatted, None)
|
||||
#
|
||||
self.commands = []
|
||||
if enable_state:
|
||||
self.commands.extend(self.STATE_COMMANDS)
|
||||
if enable_brightness:
|
||||
self.commands.extend(self.BRIGHTNESS_COMMANDS)
|
||||
if enable_color_temp:
|
||||
self.commands.extend(self.COLOR_TEMP_COMMANDS)
|
||||
self.__init_data__(enable_state, enable_brightness, enable_color_temp)
|
||||
|
||||
def __init_data__(self, enable_state, enable_brightness, enable_color_temp):
|
||||
data = {}
|
||||
if enable_state:
|
||||
data[self.KEY_OUTPUT_0] = False
|
||||
self.commands.extend(self.STATE_COMMANDS)
|
||||
if enable_brightness:
|
||||
data[self.KEY_BRIGHTNESS] = 50
|
||||
self.brightnes_move = (0, time.time())
|
||||
self.commands.extend(self.BRIGHTNESS_COMMANDS)
|
||||
if enable_color_temp:
|
||||
data[self.KEY_COLOR_TEMP] = 5
|
||||
self.commands.extend(self.COLOR_TEMP_COMMANDS)
|
||||
self.store_data(**data)
|
||||
|
||||
def __rx__(self, client, userdata, message):
|
||||
data = json.loads(message.payload)
|
||||
if self.data.get(self.KEY_OUTPUT_0) or data.get(self.KEY_OUTPUT_0) in ['on', 'toggle']:
|
||||
if message.topic.startswith(self.topic) and message.topic.endswith('/set'):
|
||||
for targetkey in data:
|
||||
value = data[targetkey]
|
||||
if targetkey in self.data.keys():
|
||||
if targetkey == self.KEY_OUTPUT_0:
|
||||
if value == "toggle":
|
||||
value = not self.data.get(self.KEY_OUTPUT_0)
|
||||
else:
|
||||
value = value == "on"
|
||||
elif targetkey == self.KEY_BRIGHTNESS:
|
||||
value = round((value - 1) / 2.53, 0)
|
||||
elif targetkey == self.KEY_COLOR_TEMP:
|
||||
value = round((value - 250) / 20.4, 0)
|
||||
self.store_data(**{targetkey: value})
|
||||
else:
|
||||
if targetkey == self.KEY_BRIGHTNESS_MOVE:
|
||||
new_value = self.data.get(self.KEY_BRIGHTNESS) + (time.time() - self.brightnes_move[1]) * self.brightnes_move[0]
|
||||
if new_value < 0:
|
||||
new_value = 0
|
||||
if new_value > 256:
|
||||
new_value = 256
|
||||
self.store_data(**{self.KEY_BRIGHTNESS: int(new_value)})
|
||||
self.brightnes_move = (value, time.time())
|
||||
else:
|
||||
print("%s: UNKNOWN KEY %s" % (message.topic, targetkey))
|
||||
elif message.topic == self.topic + '/get':
|
||||
self.__tx__(None)
|
||||
|
||||
def __tx__(self, keys_changed):
|
||||
tx_data = copy.copy(self.data)
|
||||
if self.KEY_OUTPUT_0 in tx_data:
|
||||
tx_data[self.KEY_OUTPUT_0] = "on" if tx_data[self.KEY_OUTPUT_0] else "off"
|
||||
if self.KEY_BRIGHTNESS in tx_data:
|
||||
tx_data[self.KEY_BRIGHTNESS] = 1 + round(2.53 * tx_data[self.KEY_BRIGHTNESS], 0)
|
||||
if self.KEY_COLOR_TEMP in tx_data:
|
||||
tx_data[self.KEY_COLOR_TEMP] = 250 + round(20.4 * tx_data[self.KEY_COLOR_TEMP], 0)
|
||||
self.mqtt_client.send(self.topic, json.dumps(tx_data))
|
||||
|
||||
def command(self, command):
|
||||
try:
|
||||
command, value = command.split(' ')
|
||||
except ValueError:
|
||||
value = None
|
||||
if command in self.capabilities():
|
||||
if command == self.STATE_COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_OUTPUT_0, self.data.get(self.KEY_OUTPUT_0))
|
||||
elif command == self.STATE_COMMANDS[1]:
|
||||
self.store_data(**{self.KEY_OUTPUT_0: not self.data.get(self.KEY_OUTPUT_0)})
|
||||
elif command == self.BRIGHTNESS_COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_BRIGHTNESS, self.data.get(self.KEY_BRIGHTNESS))
|
||||
elif command == self.BRIGHTNESS_COMMANDS[1]:
|
||||
self.store_data(**{self.KEY_BRIGHTNESS: command_int_value(value)})
|
||||
elif command == self.COLOR_TEMP_COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_COLOR_TEMP, self.data.get(self.KEY_COLOR_TEMP))
|
||||
elif command == self.COLOR_TEMP_COMMANDS[1]:
|
||||
self.store_data(**{self.KEY_COLOR_TEMP: command_int_value(value)})
|
||||
else:
|
||||
print("%s: not yet implemented!" % command)
|
||||
else:
|
||||
print("Unknown command!")
|
||||
|
||||
def power_off(self, device, key, value):
|
||||
self.data[self.KEY_OUTPUT_0] = False
|
||||
self.print_formatted(self, self.KEY_OUTPUT_0, False)
|
||||
|
||||
def power_on(self, device, key, value):
|
||||
if self.send_on_power_on:
|
||||
self.store_data(**{self.KEY_OUTPUT_0: True})
|
||||
else:
|
||||
self.data[self.KEY_OUTPUT_0] = True
|
||||
self.print_formatted(self, self.KEY_OUTPUT_0, True)
|
||||
|
||||
def print_formatted(self, device, key, value):
|
||||
if value is not None:
|
||||
color = COLOR_LIGHT_ACTIVE
|
||||
if key == self.KEY_OUTPUT_0:
|
||||
print_light(COLOR_LIGHT_ACTIVE, value, self.topic, "")
|
||||
self.print_formatted(device, self.KEY_BRIGHTNESS, self.data.get(self.KEY_BRIGHTNESS))
|
||||
self.print_formatted(device, self.KEY_COLOR_TEMP, self.data.get(self.KEY_COLOR_TEMP))
|
||||
elif key in [self.KEY_BRIGHTNESS, self.KEY_COLOR_TEMP]:
|
||||
perc_value = round(value, 0) if key == self.KEY_BRIGHTNESS else round(10 * value, 0)
|
||||
print_percent(
|
||||
COLOR_LIGHT_PASSIVE if not self.data.get(self.KEY_OUTPUT_0) else COLOR_LIGHT_ACTIVE,
|
||||
'B' if key == self.KEY_BRIGHTNESS else 'C',
|
||||
perc_value,
|
||||
"%3d%%" % perc_value,
|
||||
self.topic,
|
||||
""
|
||||
)
|
||||
|
||||
|
||||
class brennenstuhl_heating_valve(base):
|
||||
TEMP_RANGE = [10, 30]
|
||||
#
|
||||
KEY_TEMPERATURE_SETPOINT = "current_heating_setpoint"
|
||||
KEY_TEMPERATURE = "local_temperature"
|
||||
#
|
||||
COMMANDS = [
|
||||
"get_temperature_setpoint", "set_temperature_setpoint", "set_local_temperature",
|
||||
]
|
||||
|
||||
def __init__(self, mqtt_client, topic):
|
||||
super().__init__(mqtt_client, topic)
|
||||
self.store_data(**{
|
||||
self.KEY_TEMPERATURE_SETPOINT: 20,
|
||||
self.KEY_TEMPERATURE: 20.7,
|
||||
})
|
||||
self.add_callback(self.KEY_TEMPERATURE_SETPOINT, self.print_formatted, None)
|
||||
|
||||
def __rx__(self, client, userdata, message):
|
||||
if message.topic.startswith(self.topic) and message.topic.endswith("/set"):
|
||||
payload = payload_filter(message.payload)
|
||||
self.store_data(**payload)
|
||||
|
||||
def command(self, command):
|
||||
try:
|
||||
command, value = command.split(' ')
|
||||
except ValueError:
|
||||
value = None
|
||||
if command in self.COMMANDS:
|
||||
if command == self.COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_TEMPERATURE_SETPOINT, self.data.get(self.KEY_TEMPERATURE_SETPOINT))
|
||||
elif command == self.COMMANDS[1]:
|
||||
self.store_data(**{self.KEY_TEMPERATURE_SETPOINT: command_float_value(value)})
|
||||
elif command == self.COMMANDS[2]:
|
||||
self.store_data(**{self.KEY_TEMPERATURE: command_float_value(value)})
|
||||
|
||||
def print_formatted(self, device, key, value):
|
||||
devicename = ' - '.join(self.topic.split('/')[1:])
|
||||
if key == self.KEY_TEMPERATURE_SETPOINT:
|
||||
perc = 100 * (value - self.TEMP_RANGE[0]) / (self.TEMP_RANGE[1] - self.TEMP_RANGE[0])
|
||||
perc = 100 if perc > 100 else perc
|
||||
perc = 0 if perc < 0 else perc
|
||||
print_percent(COLOR_HEATING_VALVE, '\u03d1', perc, "%4.1f°C" % value, self.topic, "")
|
||||
|
||||
|
||||
class videv_light(base):
|
||||
AUTOSEND = False
|
||||
#
|
||||
KEY_STATE = "state"
|
||||
KEY_BRIGHTNESS = "brightness"
|
||||
KEY_COLOR_TEMP = "color_temp"
|
||||
KEY_TIMER = "timer"
|
||||
#
|
||||
STATE_COMMANDS = ("get_state", "toggle_state", )
|
||||
BRIGHTNESS_COMMANDS = ("get_brightness", "set_brightness", )
|
||||
COLOR_TEMP_COMMANDS = ("get_color_temp", "set_color_temp", )
|
||||
TIMER_COMMANDS = ("get_timer", )
|
||||
|
||||
def __init__(self, mqtt_client, topic, enable_state=True, enable_brightness=False, enable_color_temp=False, enable_timer=False):
|
||||
super().__init__(mqtt_client, topic)
|
||||
self.enable_state = enable_state
|
||||
self.enable_brightness = enable_brightness
|
||||
self.enable_color_temp = enable_color_temp
|
||||
self.enable_timer = enable_timer
|
||||
#
|
||||
self.maxvalue = None
|
||||
# add commands to be available
|
||||
if enable_state:
|
||||
# init default value
|
||||
self.data[self.KEY_STATE] = False
|
||||
# add print callback
|
||||
self.add_callback(self.KEY_STATE, self.print_formatted, None)
|
||||
# add commands to be available
|
||||
self.commands.extend(self.STATE_COMMANDS)
|
||||
if enable_brightness:
|
||||
# init default value
|
||||
self.data[self.KEY_BRIGHTNESS] = 50
|
||||
# add print callback
|
||||
self.add_callback(self.KEY_BRIGHTNESS, self.print_formatted, None)
|
||||
# add commands to be available
|
||||
self.commands.extend(self.BRIGHTNESS_COMMANDS)
|
||||
if enable_color_temp:
|
||||
# init default value
|
||||
self.data[self.KEY_COLOR_TEMP] = 5
|
||||
# add print callback
|
||||
self.add_callback(self.KEY_COLOR_TEMP, self.print_formatted, None)
|
||||
# add commands to be available
|
||||
self.commands.extend(self.COLOR_TEMP_COMMANDS)
|
||||
if enable_timer:
|
||||
# init default value
|
||||
self.data[self.KEY_TIMER] = 0
|
||||
# add print callback
|
||||
self.add_callback(self.KEY_TIMER, self.print_formatted, None)
|
||||
# add commands to be available
|
||||
self.commands.extend(self.TIMER_COMMANDS)
|
||||
|
||||
def __rx__(self, client, userdata, message):
|
||||
value = payload_filter(message.payload)
|
||||
if message.topic.startswith(self.topic):
|
||||
targetkey = message.topic.split('/')[-1]
|
||||
if targetkey in self.data.keys():
|
||||
self.store_data(**{targetkey: value})
|
||||
elif targetkey != "__info__":
|
||||
print("Unknown key %s in %s::%s" % (targetkey, message.topic, self.__class__.__name__))
|
||||
elif message.topic == self.topic + '/get':
|
||||
self.__tx__(None)
|
||||
|
||||
def send(self, key, data):
|
||||
if data is not None:
|
||||
topic = self.topic + '/' + key
|
||||
self.mqtt_client.send(topic, json.dumps(data))
|
||||
|
||||
def command(self, command):
|
||||
try:
|
||||
command, value = command.split(' ')
|
||||
except ValueError:
|
||||
value = None
|
||||
if command in self.capabilities():
|
||||
if command == self.STATE_COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_STATE, self.data.get(self.KEY_STATE))
|
||||
elif command == self.STATE_COMMANDS[1]:
|
||||
self.send(self.KEY_STATE, not self.data.get(self.KEY_STATE))
|
||||
elif command == self.BRIGHTNESS_COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_BRIGHTNESS, self.data.get(self.KEY_BRIGHTNESS))
|
||||
elif command == self.BRIGHTNESS_COMMANDS[1]:
|
||||
self.send(self.KEY_BRIGHTNESS, command_int_value(value))
|
||||
elif command == self.COLOR_TEMP_COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_COLOR_TEMP, self.data.get(self.KEY_COLOR_TEMP))
|
||||
elif command == self.COLOR_TEMP_COMMANDS[1]:
|
||||
self.send(self.KEY_COLOR_TEMP, command_int_value(value))
|
||||
elif command == self.TIMER_COMMANDS[0]:
|
||||
self.print_formatted(self, self.KEY_TIMER, self.data.get(self.KEY_TIMER))
|
||||
else:
|
||||
print("%s: not yet implemented!" % command)
|
||||
else:
|
||||
print("Unknown command!")
|
||||
|
||||
def print_formatted(self, device, key, value):
|
||||
if value is not None:
|
||||
device = " - ".join(self.topic.split('/')[1:])
|
||||
if key == self.KEY_STATE:
|
||||
print_switch(COLOR_GUI_ACTIVE, value, self.topic, "")
|
||||
elif key in [self.KEY_BRIGHTNESS, self.KEY_COLOR_TEMP]:
|
||||
perc_value = round(value * 10 if key == self.KEY_COLOR_TEMP else value, 0)
|
||||
print_percent(
|
||||
COLOR_GUI_ACTIVE,
|
||||
'B' if key == self.KEY_BRIGHTNESS else 'C',
|
||||
perc_value,
|
||||
"%3d%%" % perc_value,
|
||||
self.topic,
|
||||
""
|
||||
)
|
||||
elif key == self.KEY_TIMER:
|
||||
if value > 0:
|
||||
if self.maxvalue is None and value != 0:
|
||||
self.maxvalue = value
|
||||
disp_value = value
|
||||
try:
|
||||
perc = disp_value / self.maxvalue * 100
|
||||
except ZeroDivisionError:
|
||||
perc = 0
|
||||
else:
|
||||
disp_value = 0
|
||||
perc = 0
|
||||
self.maxvalue = None
|
||||
print_percent(COLOR_GUI_ACTIVE, 't', perc, '%3d%%' % perc, self.topic, '(%.1f)' % disp_value)
|
||||
|
||||
|
||||
# class silvercrest_motion_sensor(base):
|
||||
# KEY_OCCUPANCY = "occupancy"
|
||||
# COMMANDS = ['motion']
|
||||
|
||||
# def __init__(self, mqtt_client, topic):
|
||||
# super().__init__(mqtt_client, topic)
|
||||
# self.data[self.KEY_OCCUPANCY] = False
|
||||
# self.add_callback(self.KEY_OCCUPANCY, self.print_formatted, None)
|
||||
|
||||
# def __rx__(self, client, userdata, message):
|
||||
# pass
|
||||
|
||||
# def command(self, command):
|
||||
# try:
|
||||
# command, value = command.split(' ')
|
||||
# except ValueError:
|
||||
# value = None
|
||||
# else:
|
||||
# value = json.loads(value)
|
||||
# if command == self.COMMANDS[0]:
|
||||
# self.store_data(**{self.KEY_OCCUPANCY: True})
|
||||
# time.sleep(value or 10)
|
||||
# self.store_data(**{self.KEY_OCCUPANCY: False})
|
||||
|
||||
# def print_formatted(self, device, key, value):
|
||||
# if value is not None:
|
||||
# print_light(COLOR_MOTION_SENSOR, value, self.topic, "")
|
||||
|
||||
|
||||
# class tradfri_button(base):
|
||||
# KEY_ACTION = "action"
|
||||
# #
|
||||
# ACTION_TOGGLE = "toggle"
|
||||
# ACTION_BRIGHTNESS_UP = "brightness_up_click"
|
||||
# ACTION_BRIGHTNESS_DOWN = "brightness_down_click"
|
||||
# ACTION_RIGHT = "arrow_right_click"
|
||||
# ACTION_LEFT = "arrow_left_click"
|
||||
# ACTION_BRIGHTNESS_UP_LONG = "brightness_up_hold"
|
||||
# ACTION_BRIGHTNESS_DOWN_LONG = "brightness_down_hold"
|
||||
# ACTION_RIGHT_LONG = "arrow_right_hold"
|
||||
# ACTION_LEFT_LONG = "arrow_left_hold"
|
||||
# #
|
||||
# COMMANDS = [ACTION_TOGGLE, ACTION_LEFT, ACTION_RIGHT, ACTION_BRIGHTNESS_UP, ACTION_BRIGHTNESS_DOWN,
|
||||
# ACTION_LEFT_LONG, ACTION_RIGHT_LONG, ACTION_BRIGHTNESS_UP_LONG, ACTION_BRIGHTNESS_DOWN_LONG]
|
||||
|
||||
# def __init__(self, mqtt_client, topic):
|
||||
# super().__init__(mqtt_client, topic)
|
||||
|
||||
# def __rx__(self, client, userdata, message):
|
||||
# pass
|
||||
|
||||
# def command(self, command):
|
||||
# try:
|
||||
# command, value = command.split(' ')
|
||||
# except ValueError:
|
||||
# value = None
|
||||
# else:
|
||||
# value = json.loads(value)
|
||||
# if command in self.capabilities():
|
||||
# action = self.COMMANDS[self.COMMANDS.index(command)]
|
||||
# if self.COMMANDS.index(command) <= 4:
|
||||
# self.mqtt_client.send(self.topic, json.dumps({self.KEY_ACTION: action}))
|
||||
# elif self.COMMANDS.index(command) <= 8:
|
||||
# self.mqtt_client.send(self.topic, json.dumps({self.KEY_ACTION: action}))
|
||||
# time.sleep(value or 0.5)
|
||||
# action = '_'.join(action.split('_')[:-1] + ['release'])
|
||||
# self.mqtt_client.send(self.topic, json.dumps({self.KEY_ACTION: action}))
|
||||
|
||||
|
||||
# class remote(base):
|
||||
# def __rx__(self, client, userdata, message):
|
||||
# if message.topic == self.topic + "/VOLUP":
|
||||
# if payload_filter(message.payload):
|
||||
# icon = u'\u1403'
|
||||
# else:
|
||||
# icon = u'\u25a1'
|
||||
# elif message.topic == self.topic + "/VOLDOWN":
|
||||
# if payload_filter(message.payload):
|
||||
# icon = u'\u1401'
|
||||
# else:
|
||||
# icon = u'\u25a1'
|
||||
# else:
|
||||
# return
|
||||
# devicename = ' - '.join(self.topic.split('/')[1:-1])
|
||||
# print(COLOR_REMOTE + 10 * ' ' + icon + 6 * ' ' + devicename + colored.attr("reset"))
|
||||
|
||||
|
||||
# class gui_heating_valve(base):
|
||||
# AUTOSEND = False
|
||||
# #
|
||||
# TEMP_RANGE = [10, 30]
|
||||
# #
|
||||
# KEY_TIMER = "timer"
|
||||
# KEY_TEMPERATURE = "temperature"
|
||||
# KEY_SETPOINT_TEMP = "setpoint_temp"
|
||||
# KEY_SETPOINT_TO_DEFAULT = "setpoint_to_default"
|
||||
# KEY_BOOST = 'boost'
|
||||
# KEY_AWAY = "away"
|
||||
# KEY_SUMMER = "summer"
|
||||
# KEY_ENABLE = "enable"
|
||||
# #
|
||||
# COMMANDS = [
|
||||
# "get_temperature",
|
||||
# "get_temperature_setpoint", "set_temperature_setpoint",
|
||||
# "trigger_boost", "trigger_setpoint_to_default",
|
||||
# "toggle_away", "toggle_summer",
|
||||
# ]
|
||||
|
||||
# def __init__(self, mqtt_client, topic):
|
||||
# super().__init__(mqtt_client, topic)
|
||||
# self.add_callback(self.KEY_SETPOINT_TEMP, self.print_formatted, None)
|
||||
# self.add_callback(self.KEY_TIMER, self.print_formatted, None)
|
||||
# self.add_callback(self.KEY_AWAY, self.print_formatted, None)
|
||||
# self.add_callback(self.KEY_SUMMER, self.print_formatted, None)
|
||||
# #
|
||||
# self.store_data(**{
|
||||
# self.KEY_TEMPERATURE: 20.7,
|
||||
# self.KEY_SETPOINT_TEMP: 20,
|
||||
# self.KEY_TIMER: 0,
|
||||
# self.KEY_AWAY: False,
|
||||
# self.KEY_SUMMER: False,
|
||||
# self.KEY_ENABLE: True
|
||||
# })
|
||||
|
||||
# def __rx__(self, client, userdata, message):
|
||||
# value = payload_filter(message.payload)
|
||||
# if message.topic.startswith(self.topic) and message.topic.endswith('/set'):
|
||||
# targetkey = message.topic.split('/')[-2]
|
||||
# if targetkey in self.data.keys():
|
||||
# self.store_data(**{targetkey: value})
|
||||
# else:
|
||||
# print("Unknown key %s in %s::%s" % (targetkey, message.topic, self.__class__.__name__))
|
||||
# elif message.topic == self.topic + '/get':
|
||||
# self.__tx__(None)
|
||||
|
||||
# def send(self, key, data):
|
||||
# if data is not None:
|
||||
# topic = self.topic + '/' + key
|
||||
# self.mqtt_client.send(topic, json.dumps(data))
|
||||
|
||||
# def command(self, command):
|
||||
# try:
|
||||
# command, value = command.split(' ')
|
||||
# except ValueError:
|
||||
# value = None
|
||||
# if command in self.COMMANDS:
|
||||
# if command == self.COMMANDS[0]:
|
||||
# self.print_formatted(self, self.KEY_TEMPERATURE, self.data.get(self.KEY_TEMPERATURE))
|
||||
# elif command == self.COMMANDS[1]:
|
||||
# self.print_formatted(self, self.KEY_SETPOINT_TEMP, self.data.get(self.KEY_SETPOINT_TEMP))
|
||||
# elif command == self.COMMANDS[2]:
|
||||
# self.send(self.KEY_SETPOINT_TEMP, command_float_value(value))
|
||||
# elif command == self.COMMANDS[3]:
|
||||
# self.send(self.KEY_BOOST, True)
|
||||
# elif command == self.COMMANDS[4]:
|
||||
# self.send(self.KEY_SETPOINT_TO_DEFAULT, True)
|
||||
# elif command == self.COMMANDS[5]:
|
||||
# self.send(self.KEY_AWAY, not self.data.get(self.KEY_AWAY))
|
||||
# elif command == self.COMMANDS[6]:
|
||||
# self.send(self.KEY_SUMMER, not self.data.get(self.KEY_SUMMER))
|
||||
|
||||
# def print_formatted(self, device, key, value):
|
||||
# devicename = ' - '.join(self.topic.split('/')[1:])
|
||||
# if key == self.KEY_TIMER:
|
||||
# value /= 60
|
||||
# try:
|
||||
# perc = 100 * value / 60
|
||||
# except TypeError:
|
||||
# value = 0
|
||||
# perc = 0
|
||||
# print_percent(COLOR_GUI_ACTIVE, 't', perc, "%4.1fmin" % value, self.topic, "(Timer)")
|
||||
# elif key == self.KEY_TEMPERATURE:
|
||||
# perc = 100 * (value - self.TEMP_RANGE[0]) / (self.TEMP_RANGE[1] - self.TEMP_RANGE[0])
|
||||
# perc = 100 if perc > 100 else perc
|
||||
# perc = 0 if perc < 0 else perc
|
||||
# print_percent(COLOR_GUI_ACTIVE, '\u03d1', perc, "%4.1f°C" % value, self.topic, "(Temperature)")
|
||||
# elif key == self.KEY_SETPOINT_TEMP:
|
||||
# perc = 100 * (value - self.TEMP_RANGE[0]) / (self.TEMP_RANGE[1] - self.TEMP_RANGE[0])
|
||||
# perc = 100 if perc > 100 else perc
|
||||
# perc = 0 if perc < 0 else perc
|
||||
# print_percent(COLOR_GUI_ACTIVE if self.data.get(self.KEY_ENABLE) else COLOR_GUI_PASSIVE,
|
||||
# '\u03d1', perc, "%4.1f°C" % value, self.topic, "(Setpoint)")
|
||||
# elif key == self.KEY_AWAY:
|
||||
# print_switch(COLOR_GUI_ACTIVE, value, self.topic, "(Away Mode)")
|
||||
# elif key == self.KEY_SUMMER:
|
||||
# print_switch(COLOR_GUI_ACTIVE, value, self.topic, "(Summer Mode)")
|
||||
@ -1,263 +0,0 @@
|
||||
import config
|
||||
from __simulation__.devices import shelly, silvercrest_powerplug, tradfri_light, my_powerplug, brennenstuhl_heating_valve
|
||||
from __simulation__.devices import videv_light
|
||||
import inspect
|
||||
|
||||
|
||||
class base(object):
|
||||
def getmembers(self, prefix=''):
|
||||
rv = []
|
||||
for name, obj in inspect.getmembers(self):
|
||||
if prefix:
|
||||
full_name = prefix + '.' + name
|
||||
else:
|
||||
full_name = name
|
||||
if not name.startswith('_'):
|
||||
try:
|
||||
if obj.__module__.endswith('devices'):
|
||||
rv.append(full_name)
|
||||
else:
|
||||
rv.extend(obj.getmembers(full_name))
|
||||
except AttributeError:
|
||||
pass
|
||||
return rv
|
||||
|
||||
def getobjbyname(self, name):
|
||||
obj = self
|
||||
for subname in name.split('.'):
|
||||
obj = getattr(obj, subname)
|
||||
return obj
|
||||
|
||||
def command(self, full_command):
|
||||
try:
|
||||
parameter = " " + full_command.split(' ')[1]
|
||||
except IndexError:
|
||||
parameter = ""
|
||||
command = full_command.split(' ')[0].split('.')[-1] + parameter
|
||||
device_name = '.'.join(full_command.split(' ')[0].split('.')[:-1])
|
||||
self.getobjbyname(device_name).command(command)
|
||||
|
||||
|
||||
class gfw_floor(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_GFW_FLOOR_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
self.main_light_zigbee = tradfri_light(mqtt_client, config.TOPIC_GFW_FLOOR_MAIN_LIGHT_ZIGBEE % 1, True, True, True, False)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_on, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_off, False)
|
||||
self.main_light_zigbee_2 = tradfri_light(mqtt_client, config.TOPIC_GFW_FLOOR_MAIN_LIGHT_ZIGBEE % 2, True, True, True, False)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee_2.power_on, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee_2.power_off, False)
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_GFW_FLOOR_MAIN_LIGHT_VIDEV, True, True, True)
|
||||
|
||||
|
||||
class gfw_marion(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_GFW_MARION_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
|
||||
self.heating_valve = brennenstuhl_heating_valve(mqtt_client, config.TOPIC_GFW_MARION_HEATING_VALVE_ZIGBEE)
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_GFW_MARION_MAIN_LIGHT_VIDEV, True, False, False)
|
||||
|
||||
|
||||
class gfw_dirk(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_GFW_DIRK_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
self.main_light_zigbee = tradfri_light(mqtt_client, config.TOPIC_GFW_DIRK_MAIN_LIGHT_ZIGBEE, True, True, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_on, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_off, False)
|
||||
|
||||
self.amplifier = my_powerplug(mqtt_client, config.TOPIC_GFW_DIRK_POWERPLUG, 0)
|
||||
self.amplifier.add_channel_name(my_powerplug.KEY_OUTPUT_0, "Amplifier")
|
||||
self.desk_light = my_powerplug(mqtt_client, config.TOPIC_GFW_DIRK_POWERPLUG, 1)
|
||||
self.desk_light.add_channel_name(my_powerplug.KEY_OUTPUT_0, "Desk Light")
|
||||
self.cd_player = my_powerplug(mqtt_client, config.TOPIC_GFW_DIRK_POWERPLUG, 2)
|
||||
self.cd_player.add_channel_name(my_powerplug.KEY_OUTPUT_0, "CD_Player")
|
||||
self.pc_dock = my_powerplug(mqtt_client, config.TOPIC_GFW_DIRK_POWERPLUG, 3)
|
||||
self.pc_dock.add_channel_name(my_powerplug.KEY_OUTPUT_0, "PC_Dock")
|
||||
self.desk_light_zigbee = tradfri_light(mqtt_client, config.TOPIC_GFW_DIRK_DESK_LIGHT_ZIGBEE, True, True, True)
|
||||
self.desk_light.add_callback(my_powerplug.KEY_OUTPUT_0, self.desk_light_zigbee.power_on, True)
|
||||
self.desk_light.add_callback(my_powerplug.KEY_OUTPUT_0, self.desk_light_zigbee.power_off, False)
|
||||
|
||||
self.heating_valve = brennenstuhl_heating_valve(mqtt_client, config.TOPIC_GFW_DIRK_HEATING_VALVE_ZIGBEE)
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_GFW_DIRK_MAIN_LIGHT_VIDEV, True, True, True)
|
||||
self.videv_amplifier = videv_light(mqtt_client, config.TOPIC_GFW_DIRK_AMPLIFIER_VIDEV, True, False, False)
|
||||
self.videv_desk_light = videv_light(mqtt_client, config.TOPIC_GFW_DIRK_DESK_LIGHT_VIDEV, True, True, True)
|
||||
self.videv_cd_player = videv_light(mqtt_client, config.TOPIC_GFW_DIRK_CD_PLAYER_VIDEV, True, False, False)
|
||||
self.videv_pc_dock = videv_light(mqtt_client, config.TOPIC_GFW_DIRK_PC_DOCK_VIDEV, True, False, False)
|
||||
|
||||
|
||||
class gfw(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.floor = gfw_floor(mqtt_client)
|
||||
self.marion = gfw_marion(mqtt_client)
|
||||
self.dirk = gfw_dirk(mqtt_client)
|
||||
|
||||
|
||||
class ffw_julian(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_FFW_JULIAN_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
self.main_light_zigbee = tradfri_light(mqtt_client, config.TOPIC_FFW_JULIAN_MAIN_LIGHT_ZIGBEE, True, True, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_on, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_off, False)
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_FFW_JULIAN_MAIN_LIGHT_VIDEV, True, True, True)
|
||||
|
||||
|
||||
class ffw_livingroom(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_FFW_LIVINGROOM_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
self.main_light_zigbee = tradfri_light(mqtt_client, config.TOPIC_FFW_LIVINGROOM_MAIN_LIGHT_ZIGBEE, True, True, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_on, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_off, False)
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_FFW_LIVINGROOM_MAIN_LIGHT_VIDEV, True, True, True)
|
||||
|
||||
|
||||
class ffw_sleep(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_FFW_SLEEP_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
self.main_light_zigbee = tradfri_light(mqtt_client, config.TOPIC_FFW_SLEEP_MAIN_LIGHT_ZIGBEE, True, True, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_on, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_off, False)
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_FFW_SLEEP_MAIN_LIGHT_VIDEV, True, True, False)
|
||||
|
||||
|
||||
class ffw_bath(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.heating_valve = brennenstuhl_heating_valve(mqtt_client, config.TOPIC_FFW_BATH_HEATING_VALVE_ZIGBEE)
|
||||
|
||||
|
||||
class ffw(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.julian = ffw_julian(mqtt_client)
|
||||
self.livingroom = ffw_livingroom(mqtt_client)
|
||||
self.sleep = ffw_sleep(mqtt_client)
|
||||
self.bath = ffw_bath(mqtt_client)
|
||||
|
||||
|
||||
class ffe_floor(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_FFE_FLOOR_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_FFE_FLOOR_MAIN_LIGHT_VIDEV, True, False, False)
|
||||
|
||||
|
||||
class ffe_kitchen(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_FFE_KITCHEN_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
|
||||
self.circulation_pump = shelly(mqtt_client, config.TOPIC_FFE_KITCHEN_CIRCULATION_PUMP_SHELLY,
|
||||
input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER, output_0_auto_off=10*60)
|
||||
self.circulation_pump.add_channel_name(shelly.KEY_OUTPUT_0, "Circulation Pump")
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_FFE_KITCHEN_MAIN_LIGHT_VIDEV, True, False, False)
|
||||
self.videv_circulation_pump = videv_light(mqtt_client, config.TOPIC_FFE_KITCHEN_CIRCULATION_PUMP_VIDEV, True, False, False, True)
|
||||
|
||||
|
||||
class ffe_diningroom(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_FFE_DININGROOM_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
|
||||
self.floor_lamp = silvercrest_powerplug(mqtt_client, config.TOPIC_FFE_DININGROOM_FLOOR_LAMP_POWERPLUG)
|
||||
self.floor_lamp.add_channel_name(silvercrest_powerplug.KEY_OUTPUT_0, "Floor Lamp")
|
||||
|
||||
if config.CHRISTMAS:
|
||||
self.garland = silvercrest_powerplug(mqtt_client, config.TOPIC_FFE_DININGROOM_GARLAND_POWERPLUG)
|
||||
self.garland.add_channel_name(silvercrest_powerplug, "Garland")
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_FFE_DININGROOM_MAIN_LIGHT_VIDEV, True, False, False)
|
||||
self.videv_floor_lamp = videv_light(mqtt_client, config.TOPIC_FFE_DININGROOM_FLOOR_LAMP_VIDEV, True, False, False)
|
||||
if config.CHRISTMAS:
|
||||
self.videv_garland = videv_light(mqtt_client, config.TOPIC_FFE_DININGROOM_GARLAND_VIDEV, True, False, False)
|
||||
|
||||
|
||||
class ffe_sleep(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_FFE_SLEEP_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
self.main_light_zigbee = tradfri_light(mqtt_client, config.TOPIC_FFE_SLEEP_MAIN_LIGHT_ZIGBEE, True, True, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_on, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_off, False)
|
||||
|
||||
self.bed_light_di_zigbee = tradfri_light(mqtt_client, config.TOPIC_FFE_SLEEP_BED_LIGHT_DI_ZIGBEE, True, True, False)
|
||||
self.bed_light_ma = silvercrest_powerplug(mqtt_client, config.TOPIC_FFE_SLEEP_BED_LIGHT_MA_POWERPLUG)
|
||||
|
||||
self.heating_valve = brennenstuhl_heating_valve(mqtt_client, config.TOPIC_FFE_SLEEP_HEATING_VALVE_ZIGBEE)
|
||||
|
||||
#
|
||||
self.videv_bed_light_ma = videv_light(mqtt_client, config.TOPIC_FFE_SLEEP_BED_LIGHT_MA_VIDEV, True, False, False)
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_FFE_SLEEP_MAIN_LIGHT_VIDEV, True, True, True)
|
||||
self.videv_bed_light_di = videv_light(mqtt_client, config.TOPIC_FFE_SLEEP_BED_LIGHT_DI_VIDEV, True, True, False)
|
||||
self.videv_bed_light_ma = videv_light(mqtt_client, config.TOPIC_FFE_SLEEP_BED_LIGHT_MA_VIDEV, True, False, False)
|
||||
|
||||
|
||||
class ffe_livingroom(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_FFE_LIVINGROOM_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
self.main_light_zigbee = tradfri_light(mqtt_client, config.TOPIC_FFE_LIVINGROOM_MAIN_LIGHT_ZIGBEE, True, True, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_on, True)
|
||||
self.main_light.add_callback(shelly.KEY_OUTPUT_0, self.main_light_zigbee.power_off, False)
|
||||
|
||||
self.floor_lamp_zigbee = tradfri_light(mqtt_client, config.TOPIC_FFE_LIVINGROOM_FLOOR_LAMP_ZIGBEE % 1, True, True, True)
|
||||
for i in range(2, 7):
|
||||
setattr(self, "floor_lamp_zigbee_%d" % i, tradfri_light(mqtt_client, config.TOPIC_FFE_LIVINGROOM_FLOOR_LAMP_ZIGBEE % i, True, True, True))
|
||||
|
||||
if config.CHRISTMAS:
|
||||
self.xmas_tree = silvercrest_powerplug(mqtt_client, config.TOPIC_FFE_LIVINGROOM_XMAS_TREE_POWERPLUG)
|
||||
self.xmas_tree.add_channel_name(silvercrest_powerplug, "Xmas-Tree")
|
||||
self.xmas_star = silvercrest_powerplug(mqtt_client, config.TOPIC_FFE_LIVINGROOM_XMAS_STAR_POWERPLUG)
|
||||
self.xmas_star.add_channel_name(silvercrest_powerplug, "Xmas-Star")
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_FFE_LIVINGROOM_MAIN_LIGHT_VIDEV, True, True, True)
|
||||
self.videv_floor_lamp = videv_light(mqtt_client, config.TOPIC_FFE_LIVINGROOM_FLOOR_LAMP_VIDEV, True, True, True)
|
||||
if config.CHRISTMAS:
|
||||
self.videv_xmas_tree = videv_light(mqtt_client, config.TOPIC_FFE_LIVINGROOM_XMAS_TREE_VIDEV)
|
||||
|
||||
|
||||
class ffe(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.floor = ffe_floor(mqtt_client)
|
||||
self.kitchen = ffe_kitchen(mqtt_client)
|
||||
self.diningroom = ffe_diningroom(mqtt_client)
|
||||
self.sleep = ffe_sleep(mqtt_client)
|
||||
self.livingroom = ffe_livingroom(mqtt_client)
|
||||
|
||||
|
||||
class stairway(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.main_light = shelly(mqtt_client, config.TOPIC_STW_STAIRWAY_MAIN_LIGHT_SHELLY, input_0_func=shelly.INPUT_FUNC_OUT1_TRIGGER)
|
||||
self.main_light.add_channel_name(shelly.KEY_OUTPUT_0, "Main Light")
|
||||
|
||||
#
|
||||
self.videv_main_light = videv_light(mqtt_client, config.TOPIC_STW_STAIRWAY_MAIN_LIGHT_VIDEV, True, False, False, True)
|
||||
|
||||
|
||||
class house(base):
|
||||
def __init__(self, mqtt_client):
|
||||
self.gfw = gfw(mqtt_client)
|
||||
self.ffw = ffw(mqtt_client)
|
||||
self.ffe = ffe(mqtt_client)
|
||||
self.stairway = stairway(mqtt_client)
|
||||
@ -1,249 +0,0 @@
|
||||
import colored
|
||||
import inspect
|
||||
from __simulation__ import devices
|
||||
import time
|
||||
|
||||
|
||||
DT_TOGGLE = 0.3
|
||||
|
||||
|
||||
TEST_FULL = 'full'
|
||||
TEST_SMOKE = 'smoke'
|
||||
#
|
||||
COLOR_SUCCESS = colored.fg("light_green")
|
||||
COLOR_FAIL = colored.fg("light_red")
|
||||
|
||||
|
||||
class test_smarthome(object):
|
||||
def __init__(self, rooms):
|
||||
# add testcases by room objects
|
||||
for name in rooms.getmembers():
|
||||
obj = rooms.getobjbyname(name)
|
||||
if obj.__class__.__name__ == "videv_light":
|
||||
common_name = '.'.join(name.split('.')[:-1]) + '.' + name.split('.')[-1][6:]
|
||||
li_device = rooms.getobjbyname(common_name + '_zigbee') if obj.enable_brightness or obj.enable_color_temp else None
|
||||
try:
|
||||
sw_device = rooms.getobjbyname(common_name) if obj.enable_state else None
|
||||
except AttributeError:
|
||||
# must be a device without switching device
|
||||
sw_device = li_device
|
||||
setattr(self, common_name.replace('.', '_'), testcase_light(obj, sw_device, li_device))
|
||||
# add test collection
|
||||
self.all = test_collection(self)
|
||||
|
||||
def getmembers(self, prefix=''):
|
||||
rv = []
|
||||
for name, obj in inspect.getmembers(self):
|
||||
if prefix:
|
||||
full_name = prefix + '.' + name
|
||||
else:
|
||||
full_name = name
|
||||
if not name.startswith('_'):
|
||||
try:
|
||||
if obj.__class__.__bases__[0].__name__ == "testcase" or obj.__class__.__name__ == "test_collection":
|
||||
rv.append(full_name)
|
||||
else:
|
||||
rv.extend(obj.getmembers(full_name))
|
||||
except AttributeError:
|
||||
pass
|
||||
return rv
|
||||
|
||||
def getobjbyname(self, name):
|
||||
if name.startswith("test."):
|
||||
name = name[5:]
|
||||
obj = self
|
||||
for subname in name.split('.'):
|
||||
obj = getattr(obj, subname)
|
||||
return obj
|
||||
|
||||
def command(self, full_command):
|
||||
try:
|
||||
parameter = " " + full_command.split(' ')[1]
|
||||
except IndexError:
|
||||
parameter = ""
|
||||
command = full_command.split(' ')[0].split('.')[-1] + parameter
|
||||
device_name = '.'.join(full_command.split(' ')[0].split('.')[:-1])
|
||||
self.getobjbyname(device_name).command(command)
|
||||
|
||||
|
||||
class test_result_base(object):
|
||||
def __init__(self):
|
||||
self.__init_test_counters__()
|
||||
|
||||
def __init_test_counters__(self):
|
||||
self.test_counter = 0
|
||||
self.success_tests = 0
|
||||
self.failed_tests = 0
|
||||
|
||||
def statistic(self):
|
||||
return (self.test_counter, self.success_tests, self.failed_tests)
|
||||
|
||||
def print_statistic(self):
|
||||
color = COLOR_SUCCESS if self.test_counter == self.success_tests else COLOR_FAIL
|
||||
print(color + "*** SUCCESS: (%4d/%4d) FAIL: (%4d/%4d) ***\n" % (self.success_tests,
|
||||
self.test_counter, self.failed_tests, self.test_counter) + colored.attr("reset"))
|
||||
|
||||
|
||||
class test_collection(test_result_base):
|
||||
def __init__(self, test_instance):
|
||||
super().__init__()
|
||||
self.test_instance = test_instance
|
||||
|
||||
def capabilities(self):
|
||||
return [TEST_FULL, TEST_SMOKE]
|
||||
|
||||
def command(self, command):
|
||||
self.__init_test_counters__()
|
||||
for member in self.test_instance.getmembers():
|
||||
obj = self.test_instance.getobjbyname(member)
|
||||
if id(obj) != id(self):
|
||||
obj.test_all(command)
|
||||
num, suc, fail = obj.statistic()
|
||||
self.test_counter += num
|
||||
self.success_tests += suc
|
||||
self.failed_tests += fail
|
||||
self.print_statistic()
|
||||
|
||||
|
||||
class testcase(test_result_base):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.__test_list__ = []
|
||||
|
||||
def capabilities(self):
|
||||
if len(self.__test_list__) > 0 and not 'test_all' in self.__test_list__:
|
||||
self.__test_list__.append('test_all')
|
||||
self.__test_list__.sort()
|
||||
return self.__test_list__
|
||||
|
||||
def test_all(self, test=TEST_FULL):
|
||||
test_counter = 0
|
||||
success_tests = 0
|
||||
failed_tests = 0
|
||||
for tc_name in self.capabilities():
|
||||
if tc_name != "test_all":
|
||||
self.command(tc_name, test)
|
||||
test_counter += self.test_counter
|
||||
success_tests += self.success_tests
|
||||
failed_tests += self.failed_tests
|
||||
self.test_counter = test_counter
|
||||
self.success_tests = success_tests
|
||||
self.failed_tests = failed_tests
|
||||
|
||||
def command(self, command, test=TEST_FULL):
|
||||
self.__init_test_counters__()
|
||||
tc = getattr(self, command)
|
||||
self.__init_test_counters__()
|
||||
rv = tc(test)
|
||||
self.print_statistic()
|
||||
|
||||
def heading(self, desciption):
|
||||
print(desciption)
|
||||
|
||||
def sub_heading(self, desciption):
|
||||
print(2 * " " + desciption)
|
||||
|
||||
def result(self, desciption, success):
|
||||
self.test_counter += 1
|
||||
if success:
|
||||
self.success_tests += 1
|
||||
else:
|
||||
self.failed_tests += 1
|
||||
print(4 * " " + ("SUCCESS - " if success else "FAIL - ") + desciption)
|
||||
|
||||
|
||||
class testcase_light(testcase):
|
||||
def __init__(self, videv, sw_device, li_device):
|
||||
self.videv = videv
|
||||
self.sw_device = sw_device
|
||||
self.li_device = li_device
|
||||
self.__test_list__ = []
|
||||
if self.videv.enable_state:
|
||||
self.__test_list__.append('test_power_on_off')
|
||||
if self.videv.enable_brightness:
|
||||
self.__test_list__.append('test_brightness')
|
||||
if self.videv.enable_color_temp:
|
||||
self.__test_list__.append('test_color_temp')
|
||||
|
||||
def test_power_on_off(self, test=TEST_FULL):
|
||||
self.heading("Power On/ Off test (%s)" % self.videv.topic)
|
||||
#
|
||||
sw_state = self.sw_device.get_data(self.sw_device.KEY_OUTPUT_0)
|
||||
#
|
||||
for i in range(0, 2):
|
||||
self.sub_heading("State change of switching device")
|
||||
#
|
||||
self.sw_device.store_data(**{self.sw_device.KEY_OUTPUT_0: not self.sw_device.data.get(self.sw_device.KEY_OUTPUT_0)})
|
||||
time.sleep(DT_TOGGLE)
|
||||
self.result("Virtual device state after Switch on by switching device", sw_state != self.videv.get_data(self.videv.KEY_STATE))
|
||||
self.result("Switching device state after Switch on by switching device",
|
||||
sw_state != self.sw_device.get_data(self.sw_device.KEY_OUTPUT_0))
|
||||
|
||||
self.sub_heading("State change of virtual device")
|
||||
#
|
||||
self.videv.send(self.videv.KEY_STATE, not self.videv.data.get(self.videv.KEY_STATE))
|
||||
time.sleep(DT_TOGGLE)
|
||||
self.result("Virtual device state after Switch off by virtual device", sw_state == self.videv.get_data(self.videv.KEY_STATE))
|
||||
self.result("Switching device state after Switch on by switching device",
|
||||
sw_state == self.sw_device.get_data(self.sw_device.KEY_OUTPUT_0))
|
||||
|
||||
def test_brightness(self, test=TEST_FULL):
|
||||
self.heading("Brightness test (%s)" % self.videv.topic)
|
||||
#
|
||||
br_state = self.li_device.get_data(self.li_device.KEY_BRIGHTNESS)
|
||||
delta = -15 if br_state > 50 else 15
|
||||
|
||||
self.sw_device.store_data(**{self.sw_device.KEY_OUTPUT_0: True})
|
||||
time.sleep(DT_TOGGLE)
|
||||
|
||||
for i in range(0, 2):
|
||||
self.sub_heading("Brightness change by light device")
|
||||
#
|
||||
self.li_device.store_data(**{self.li_device.KEY_BRIGHTNESS: br_state + delta})
|
||||
time.sleep(DT_TOGGLE)
|
||||
self.result("Virtual device state after setting brightness by light device",
|
||||
br_state + delta == self.videv.get_data(self.videv.KEY_BRIGHTNESS))
|
||||
self.result("Light device state after setting brightness by light device", br_state +
|
||||
delta == self.li_device.get_data(self.li_device.KEY_BRIGHTNESS))
|
||||
|
||||
self.sub_heading("Brightness change by virtual device")
|
||||
#
|
||||
self.videv.send(self.videv.KEY_BRIGHTNESS, br_state)
|
||||
time.sleep(DT_TOGGLE)
|
||||
self.result("Virtual device state after setting brightness by light device", br_state == self.videv.get_data(self.videv.KEY_BRIGHTNESS))
|
||||
self.result("Light device state after setting brightness by light device",
|
||||
br_state == self.li_device.get_data(self.li_device.KEY_BRIGHTNESS))
|
||||
|
||||
self.sw_device.store_data(**{self.sw_device.KEY_OUTPUT_0: False})
|
||||
time.sleep(DT_TOGGLE)
|
||||
|
||||
def test_color_temp(self, test=TEST_FULL):
|
||||
self.heading("Color temperature test (%s)" % self.videv.topic)
|
||||
#
|
||||
ct_state = self.li_device.get_data(self.li_device.KEY_COLOR_TEMP)
|
||||
delta = -3 if ct_state > 5 else 3
|
||||
|
||||
self.sw_device.store_data(**{self.sw_device.KEY_OUTPUT_0: True})
|
||||
time.sleep(DT_TOGGLE)
|
||||
|
||||
for i in range(0, 2):
|
||||
self.sub_heading("Color temperature change by light device")
|
||||
#
|
||||
self.li_device.store_data(**{self.li_device.KEY_COLOR_TEMP: ct_state + delta})
|
||||
time.sleep(DT_TOGGLE)
|
||||
self.result("Virtual device state after setting color temperature by light device",
|
||||
ct_state + delta == self.videv.get_data(self.videv.KEY_COLOR_TEMP))
|
||||
self.result("Light device state after setting color temperature by light device", ct_state +
|
||||
delta == self.li_device.get_data(self.li_device.KEY_COLOR_TEMP))
|
||||
|
||||
self.sub_heading("Color temperature change by virtual device")
|
||||
#
|
||||
self.videv.send(self.videv.KEY_COLOR_TEMP, ct_state)
|
||||
time.sleep(DT_TOGGLE)
|
||||
self.result("Virtual device state after setting color temperature by light device",
|
||||
ct_state == self.videv.get_data(self.videv.KEY_COLOR_TEMP))
|
||||
self.result("Light device state after setting color temperature by light device",
|
||||
ct_state == self.li_device.get_data(self.li_device.KEY_COLOR_TEMP))
|
||||
|
||||
self.sw_device.store_data(**{self.sw_device.KEY_OUTPUT_0: False})
|
||||
time.sleep(DT_TOGGLE)
|
||||
@ -1,84 +0,0 @@
|
||||
import config
|
||||
import logging
|
||||
import mqtt
|
||||
import readline
|
||||
import report
|
||||
from __simulation__.rooms import house
|
||||
from __simulation__.test import test_smarthome
|
||||
import time
|
||||
|
||||
if __name__ == "__main__":
|
||||
report.stdoutLoggingConfigure(((config.APP_NAME, logging.WARNING), ), 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 + '_simulation')
|
||||
#
|
||||
COMMANDS = ['quit', 'help']
|
||||
#
|
||||
h = house(mc)
|
||||
for name in h.getmembers():
|
||||
d = h.getobjbyname(name)
|
||||
for c in d.capabilities():
|
||||
COMMANDS.append(name + '.' + c)
|
||||
#
|
||||
ts = test_smarthome(h)
|
||||
for name in ts.getmembers():
|
||||
d = ts.getobjbyname(name)
|
||||
for c in d.capabilities():
|
||||
COMMANDS.append('test.' + name + '.' + c)
|
||||
|
||||
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
|
||||
|
||||
readline.parse_and_bind("tab: complete")
|
||||
readline.set_completer(completer)
|
||||
time.sleep(0.3)
|
||||
print("\nEnter command: ")
|
||||
|
||||
while True:
|
||||
userfeedback = input('')
|
||||
command = userfeedback.split(' ')[0]
|
||||
if userfeedback == 'quit':
|
||||
break
|
||||
elif userfeedback == 'help':
|
||||
print("Help is not yet implemented!")
|
||||
elif userfeedback.startswith("test"):
|
||||
ts.command(userfeedback)
|
||||
elif userfeedback == 'test.smoke':
|
||||
ts.smoke()
|
||||
elif command in COMMANDS[2:]:
|
||||
h.command(userfeedback)
|
||||
elif userfeedback != "":
|
||||
print("Unknown command!")
|
||||
else:
|
||||
print()
|
||||
Loading…
x
Reference in New Issue
Block a user