diff --git a/__simulation__/__init__.py b/__simulation__/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/__simulation__/devices.py b/__simulation__/devices.py deleted file mode 100644 index d0a90b7..0000000 --- a/__simulation__/devices.py +++ /dev/null @@ -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)") diff --git a/__simulation__/rooms.py b/__simulation__/rooms.py deleted file mode 100644 index 1663fc9..0000000 --- a/__simulation__/rooms.py +++ /dev/null @@ -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) diff --git a/__simulation__/test.py b/__simulation__/test.py deleted file mode 100644 index 1350fb8..0000000 --- a/__simulation__/test.py +++ /dev/null @@ -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) diff --git a/house_n_gui_sim.py b/house_n_gui_sim.py deleted file mode 100644 index 696c6af..0000000 --- a/house_n_gui_sim.py +++ /dev/null @@ -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()