smarthome/smart_brain
1
0
Förgrening 0
Kod Ärenden Pull-förfrågningar Släpp Wiki Aktiviteter

simulation for circulation pump, heating valve and amplifier remote added

Bläddra i källkod
This commit is contained in:
Dirk Alders 2023-01-06 21:41:15 +01:00
förälder 46a2c598ae
incheckning 03a6f83c42
2 ändrade filer med 235 tillägg och 45 borttagningar
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

259
__simulation__/devices.py
Visa fil

@ -1,6 +1,7 @@
import colored
import json
import sys
import task
import time
COLOR_GUI_ACTIVE = colored.fg("light_blue")
@ -9,7 +10,9 @@ 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("red")
COLOR_MOTION_SENSOR = colored.fg("dark_orange_3b")
COLOR_RADIATOR_VALVE = colored.fg("red")
COLOR_REMOTE = colored.fg("green")
def payload_filter(payload):
@ -19,12 +22,19 @@ def payload_filter(payload):
return payload.decode("utf-8")
def percent_print(value):
def percent_bar(value):
rv = ""
for i in range(0, 10):
if (value - 5) > 10*i:
sys.stdout.write(u"\u25ac")
else:
sys.stdout.write(u"\u25ad")
rv += u"\u25ac" if (value - 5) > 10*i else u"\u25ad"
return rv
def green_led():
return colored.fg('green') + "\u2b24" + colored.attr("reset")
def grey_led():
return colored.fg('light_gray') + "\u2b24" + colored.attr("reset")
def command_int_value(value):
@ -34,6 +44,13 @@ def command_int_value(value):
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))
class base(object):
AUTOSEND = True
COMMANDS = []
@ -96,19 +113,24 @@ class shelly(base):
COMMANDS = [
"get_relay_0", "toggle_relay_0",
"get_relay_1", "toggle_relay_1",
"get_input_0", "trigger_input_0",
"get_input_1", "trigger_input_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):
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: "off", self.KEY_OUTPUT_1: "off", self.KEY_INPUT_0: "off", self.KEY_INPUT_1: "off"})
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__, "on")
self.add_callback(self.KEY_OUTPUT_0, self.__stop_auto_off__, "off")
self.add_callback(self.KEY_OUTPUT_1, self.print_formatted, None)
#
self.add_callback(self.KEY_INPUT_0, self.__input_function__, None)
@ -136,14 +158,27 @@ class shelly(base):
func = None
if func == self.INPUT_FUNC_OUT1_FOLLOW:
self.__set_data__(self.KEY_OUTPUT_0, data)
elif func == self.INPUT_FUNC_OUT1_TRIGGER and data == 'on':
print("Triggered Output 0 by Input 0")
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 and data == 'on':
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):
if value in ["on", "off"]:
self.store_data(**{key: value})
@ -167,29 +202,23 @@ class shelly(base):
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.__set_data__(self.KEY_INPUT_0, 'on')
time.sleep(0.2)
self.__set_data__(self.KEY_INPUT_0, 'off')
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.__set_data__(self.KEY_INPUT_1, 'on')
time.sleep(0.2)
self.__set_data__(self.KEY_INPUT_1, 'off')
self.__toggle_data__(self.KEY_INPUT_1)
elif command == self.COMMANDS[8]:
self.__set_data__(self.KEY_INPUT_0, 'on')
self.__toggle_data__(self.KEY_INPUT_0)
time.sleep(0.4)
self.__set_data__(self.KEY_LONGPUSH_0, 'on')
time.sleep(0.1)
self.__set_data__(self.KEY_LONGPUSH_0, 'off')
self.__set_data__(self.KEY_INPUT_0, 'off')
elif command == self.COMMANDS[9]:
self.__set_data__(self.KEY_INPUT_1, 'on')
self.__toggle_data__(self.KEY_INPUT_1)
time.sleep(0.4)
self.__set_data__(self.KEY_LONGPUSH_1, 'on')
time.sleep(0.1)
self.__set_data__(self.KEY_LONGPUSH_1, 'off')
self.__set_data__(self.KEY_INPUT_1, 'off')
else:
print("%s: not yet implemented!" % command)
else:
@ -198,9 +227,10 @@ class shelly(base):
def print_formatted(self, device, key, value):
if value is not None:
icon = u'\u2b24' if value == "on" else u'\u25ef'
channel = "(%s)" % self.names.get(key, key)
info = (" - %ds" % self.__output_0_auto_off__) if self.__output_0_auto_off__ is not None and value == "on" else ""
channel = "(%s%s)" % (self.names.get(key, key), info)
devicename = " - ".join(self.topic.split('/')[1:])
print(COLOR_SHELLY + 10 * ' ' + icon + 6 * ' ' + devicename + ' ' + channel + colored.attr("reset"))
print(COLOR_SHELLY + 10 * ' ' + icon + 9 * ' ' + devicename + ' ' + channel + colored.attr("reset"))
class my_powerplug(base):
@ -230,7 +260,6 @@ class my_powerplug(base):
channels = range(0, 4)
for channel in channels:
payload = payload_filter(message.payload)
print(channel, payload)
if payload == "toggle":
payload = not self.data.get(str(channel))
self.store_data(**{str(channel): payload})
@ -267,7 +296,7 @@ class my_powerplug(base):
icon = u'\u2b24' if value else u'\u25ef'
channel = "(%s)" % self.names.get(key, "Channel %d" % (int(key) + 1))
devicename = " - ".join(self.topic.split('/')[1:])
print(COLOR_POWERPLUG + 10 * ' ' + icon + 6 * ' ' + devicename + ' ' + channel + colored.attr("reset"))
print(COLOR_POWERPLUG + 10 * ' ' + icon + 9 * ' ' + devicename + ' ' + channel + colored.attr("reset"))
class silvercrest_powerplug(base):
@ -314,7 +343,7 @@ class silvercrest_powerplug(base):
icon = u'\u2b24' if value == "on" else u'\u25ef'
channel = "(%s)" % self.names.get(key, key)
devicename = " - ".join(self.topic.split('/')[1:])
print(COLOR_POWERPLUG + 10 * ' ' + icon + 6 * ' ' + devicename + ' ' + channel + colored.attr("reset"))
print(COLOR_POWERPLUG + 10 * ' ' + icon + 9 * ' ' + devicename + ' ' + channel + colored.attr("reset"))
class silvercrest_motion_sensor(base):
@ -344,9 +373,9 @@ class silvercrest_motion_sensor(base):
def print_formatted(self, device, key, value):
if value is not None:
if value:
print(COLOR_MOTION_SENSOR + 10 * ' ' + u'\u2b24' + 6 * ' ' + " - ".join(self.topic.split('/')[1:]) + colored.attr("reset"))
print(COLOR_MOTION_SENSOR + 10 * ' ' + u'\u2b24' + 9 * ' ' + " - ".join(self.topic.split('/')[1:]) + colored.attr("reset"))
else:
print(COLOR_MOTION_SENSOR + 10 * ' ' + u'\u25ef' + 6 * ' ' + " - ".join(self.topic.split('/')[1:]) + colored.attr("reset"))
print(COLOR_MOTION_SENSOR + 10 * ' ' + u'\u25ef' + 9 * ' ' + " - ".join(self.topic.split('/')[1:]) + colored.attr("reset"))
class tradfri_light(base):
@ -452,9 +481,9 @@ class tradfri_light(base):
color = COLOR_LIGHT_ACTIVE
if key == self.KEY_STATE:
if value == "on":
print(color + 10 * ' ' + u'\u2b24' + 6 * ' ' + " - ".join(self.topic.split('/')[1:]) + colored.attr("reset"))
print(color + 10 * ' ' + u'\u2b24' + 9 * ' ' + " - ".join(self.topic.split('/')[1:]) + colored.attr("reset"))
else:
print(color + 10 * ' ' + u'\u25ef' + 6 * ' ' + " - ".join(self.topic.split('/')[1:]) + colored.attr("reset"))
print(color + 10 * ' ' + u'\u25ef' + 9 * ' ' + " - ".join(self.topic.split('/')[1:]) + colored.attr("reset"))
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]:
@ -466,8 +495,8 @@ class tradfri_light(base):
value = round((value - 250) * 100 / 204, 0)
sys.stdout.write(color)
sys.stdout.write('B' if key == gui_light.KEY_BRIGHTNESS else 'C')
percent_print(value)
sys.stdout.write("%3d%% " % value)
sys.stdout.write(percent_bar(value))
sys.stdout.write("%3d%%" % value + 5 * " ")
print(" - ".join(self.topic.split('/')[1:]) + colored.attr("reset"))
@ -535,9 +564,9 @@ class gui_light(tradfri_light):
color = COLOR_GUI_ACTIVE
if key == self.KEY_STATE:
if value == True:
print(color + 10 * ' ' + u'\u25a0' + 6 * ' ' + " - ".join(self.topic.split('/')[1:-1]) + colored.attr("reset"))
print(color + 10 * ' ' + u'\u25a0' + 9 * ' ' + " - ".join(self.topic.split('/')[1:-1]) + colored.attr("reset"))
else:
print(color + 10 * ' ' + u'\u25a1' + 6*' ' + " - ".join(self.topic.split('/')[1:-1]) + colored.attr("reset"))
print(color + 10 * ' ' + u'\u25a1' + 9*' ' + " - ".join(self.topic.split('/')[1:-1]) + colored.attr("reset"))
elif key == self.KEY_ENABLE:
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))
@ -547,8 +576,8 @@ class gui_light(tradfri_light):
value = round(value * 10 if key == self.KEY_COLOR_TEMP else value, 0)
sys.stdout.write(color)
sys.stdout.write('B' if key == self.KEY_BRIGHTNESS else 'C')
percent_print(value)
sys.stdout.write("%3d%% " % value)
sys.stdout.write(percent_bar(value))
sys.stdout.write("%3d%%" % value + 5 * " ")
print(" - ".join(self.topic.split('/')[1:-1]) + colored.attr("reset"))
@ -623,7 +652,159 @@ class gui_led_array(base):
print("Unknown key %s in %s" % (targetkey, self.__class__.__name__))
def print_formatted(self, device, key, value):
color = colored.fg('green') if value else ""
led = green_led() if value else grey_led()
channel = '(%s)' % self.names.get(key, key)
devicename = ' - '.join(self.topic.split('/')[1:-1])
print(color + 10 * ' ' + u"\u2b24" + 6 * ' ' + COLOR_GUI_ACTIVE + devicename + ' ' + channel + colored.attr("reset"))
print(10 * ' ' + led + 9 * ' ' + COLOR_GUI_ACTIVE + devicename + ' ' + channel + colored.attr("reset"))
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_timer(base):
AUTOSEND = False
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
self.maxvalue = None
self.last_printed = None
def __rx__(self, client, userdata, message):
payload = payload_filter(message.payload)
if message.topic.startswith(self.topic) and message.topic.endswith('/feedback/set'):
if isinstance(payload, (int, float, complex)) and not isinstance(payload, bool):
if self.maxvalue is None:
self.maxvalue = payload
perc = payload / self.maxvalue * 100
if self.last_printed is None or abs(self.last_printed - perc) >= 4.8:
sys.stdout.write(COLOR_GUI_ACTIVE + 't' + percent_bar(perc))
print('%3d%%' % perc + 2 * ' ' + ' - '.join(self.topic.split('/')[1:-1]) + ' (%.1f)' % payload + colored.attr('reset'))
self.last_printed = perc
else:
self.maxvalue = None
self.last_printed = None
sys.stdout.write(COLOR_GUI_ACTIVE + 't' + percent_bar(0))
print('%3d%%' % 0 + 2 * ' ' + ' - '.join(self.topic.split('/')[1:-1]) + colored.attr('reset'))
else:
print("Unknown Message")
class brennenstuhl_radiator_valve(base):
TEMP_RANGE = [10, 30]
#
KEY_TEMPERATURE_SETPOINT = "current_heating_setpoint"
KEY_TEMPERATURE = "local_temperature"
#
COMMANDS = [
"get_temperature_setpoint", "set_temperature_setpoint",
]
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)})
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
sys.stdout.write(COLOR_RADIATOR_VALVE + '\u03d1' + percent_bar(perc))
sys.stdout.write("%4.1f°C" % value + 3 * " ")
print(devicename + colored.attr("reset"))
class gui_radiator_valve(base):
AUTOSEND = False
#
TEMP_RANGE = [10, 30]
#
KEY_BOOST_LED = "boost_led"
KEY_TEMPERATURE_SETPOINT = "temperature_setpoint"
#
COMMANDS = [
"get_temperature_setpoint", "set_temperature_setpoint",
"get_boost_state", "trigger_boost",
]
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
self.add_callback(self.KEY_TEMPERATURE_SETPOINT, self.print_formatted, None)
self.add_callback(self.KEY_BOOST_LED, 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)
if type(payload) == bool:
self.store_data(**{self.KEY_BOOST_LED: payload})
else:
self.store_data(**{self.KEY_TEMPERATURE_SETPOINT: payload})
else:
print(message.topic)
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]:
################################### TEMPORARY!!! ###################################
self.mqtt_client.send(self.topic + "/feedback/set", command_float_value(value))
################################### TEMPORARY!!! ###################################
elif command == self.COMMANDS[2]:
self.print_formatted(self, self.KEY_BOOST_LED, self.data.get(self.KEY_BOOST_LED))
elif command == self.COMMANDS[3]:
################################### TEMPORARY!!! ###################################
self.mqtt_client.send(self.topic + "/state", json.dumps(True))
################################### TEMPORARY!!! ###################################
def print_formatted(self, device, key, value):
devicename = ' - '.join(self.topic.split('/')[1:])
if key == self.KEY_BOOST_LED:
led = green_led() if value else grey_led()
print(10 * ' ' + led + 9 * ' ' + COLOR_GUI_ACTIVE + devicename + " (Boost)" + colored.attr("reset"))
elif 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
sys.stdout.write(COLOR_GUI_ACTIVE + '\u03d1' + percent_bar(perc))
sys.stdout.write("%4.1f°C" % value + 3 * " ")
print(devicename + colored.attr("reset"))

21
__simulation__/rooms.py
Visa fil

@ -1,12 +1,8 @@
import config
from __simulation__.devices import shelly, silvercrest_powerplug, gui_light, tradfri_light, tradfri_button, gui_led_array, silvercrest_motion_sensor, my_powerplug
from __simulation__.devices import shelly, silvercrest_powerplug, tradfri_light, tradfri_button, silvercrest_motion_sensor, my_powerplug, remote, brennenstuhl_radiator_valve
from __simulation__.devices import gui_light, gui_led_array, gui_timer, gui_radiator_valve
import inspect
# TODO: ffe_sleep: Implement heating valve
# TODO: gfw_dirk: Add at least brightness amplifier remote feedback to console
# TODO: ffe_diningroom: Implement garland gui_switch
# TODO: ffe_kitchen: Implement circulation pump (incl. shelly auto off function)
class base(object):
def getmembers(self, prefix=''):
@ -84,6 +80,8 @@ class gfw_dirk(base):
self.gui_switch_cd_player = gui_light(mqtt_client, config.TOPIC_GFW_DIRK_CD_PLAYER_GUI_SWITCH, True, False, False)
self.gui_switch_pc_dock = gui_light(mqtt_client, config.TOPIC_GFW_DIRK_PC_DOCK_GUI_SWITCH, True, False, False)
#
self.remote = remote(mqtt_client, config.TOPIC_GFW_DIRK_AMPLIFIER_REMOTE)
#
self.input_device = tradfri_button(mqtt_client, config.TOPIC_GFW_DIRK_INPUT_DEVICE)
self.led_array = gui_led_array(mqtt_client, config.TOPIC_GFW_DIRK_DEVICE_CHOOSER_LED)
self.led_array.add_channel_name(gui_led_array.KEY_LED_0, "Main Light")
@ -134,6 +132,12 @@ class ffe_kitchen(base):
self.gui_switch_main_light = gui_light(mqtt_client, config.TOPIC_FFE_KITCHEN_MAIN_LIGHT_GUI_SWITCH, True, False, False)
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.gui_switch_circulation_pump = gui_light(mqtt_client, config.TOPIC_FFE_KITCHEN_CIRCULATION_PUMP_GUI_SWITCH, True, False, False)
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.gui_timer_circulation_pump = gui_timer(mqtt_client, config.TOPIC_FFE_KITCHEN_CIRCULATION_PUMP_GUI_TIMER)
class ffe_diningroom(base):
@ -166,6 +170,11 @@ class ffe_sleep(base):
self.led_array = gui_led_array(mqtt_client, config.TOPIC_FFE_SLEEP_DEVICE_CHOOSER_LED)
self.led_array.add_channel_name(gui_led_array.KEY_LED_0, "Main Light")
self.led_array.add_channel_name(gui_led_array.KEY_LED_1, "Bed Light Dirk")
#
self.heating_valve = brennenstuhl_radiator_valve(mqtt_client, config.TOPIC_FFE_SLEEP_RADIATOR_VALVE_ZIGBEE)
self.gui_heating_valve = gui_radiator_valve(mqtt_client, "gui/ffe_ts_sleep_madi")
self.gui_heating_valve_boost_led = gui_radiator_valve(mqtt_client, "gui/ffe_bl_sleep_madi")
self.gui_heating_valve_boost_button = gui_radiator_valve(mqtt_client, "gui/ffe_bo_sleep_madi")
class ffe_livingroom(base):