smarthome/smart_enlife
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

First functional version

Browse Source
This commit is contained in:
Dirk Alders 2024-12-09 20:21:10 +01:00
parent 31acf5db99
commit f4c8eebc98
19 changed files with 1423 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
.gitmodules vendored Normal file
View File

@ -0,0 +1,15 @@
[submodule "mqtt"]
path = mqtt
url = https://git.mount-mockery.de/pylib/mqtt.git
[submodule "report"]
path = report
url = https://git.mount-mockery.de/pylib/report.git
[submodule "geo"]
path = geo
url = https://git.mount-mockery.de/pylib/geo.git
[submodule "task"]
path = task
url = https://git.mount-mockery.de/pylib/task.git
[submodule "state_machine"]
path = state_machine
url = https://git.mount-mockery.de/pylib/state_machine.git

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

@ -0,0 +1,16 @@
{
// Verwendet IntelliSense zum Ermitteln möglicher Attribute.
// Zeigen Sie auf vorhandene Attribute, um die zugehörigen Beschreibungen anzuzeigen.
// Weitere Informationen finden Sie unter https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Python: Main File execution",
"type": "debugpy",
"request": "launch",
"program": "${workspaceFolder}/smart_enlife.py",
"console": "integratedTerminal",
"justMyCode": true
}
]
}

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

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

138
base.py Normal file
View File

@ -0,0 +1,138 @@
import json
import logging
import task
try:
from config import APP_NAME as ROOT_LOGGER_NAME
except ImportError:
ROOT_LOGGER_NAME = 'root'
class common_base(dict):
DEFAULT_VALUES = {}
def __init__(self, default_values=None):
super().__init__(default_values or self.DEFAULT_VALUES)
#
self.__callback_list__ = []
self.logger = logging.getLogger(ROOT_LOGGER_NAME).getChild("devices")
def add_callback(self, key, data, callback, on_change_only=False, init_now=False):
"""
key: key or None for all keys
data: data or None for all data
"""
cb_tup = (key, data, callback, on_change_only)
if cb_tup not in self.__callback_list__:
self.__callback_list__.append(cb_tup)
if init_now and self.get(key) is not None:
callback(self, key, self[key])
def set(self, key, data, block_callback=[]):
if key in self.keys():
value_changed = self[key] != data
self[key] = data
for cb_key, cb_data, callback, on_change_only in self.__callback_list__:
if cb_key is None or key == cb_key: # key fits to callback definition
if cb_data is None or cb_data == self[key]: # data fits to callback definition
if value_changed or not on_change_only: # change status fits to callback definition
if not callback in block_callback: # block given callbacks
callback(self, key, self[key])
else:
self.logger.warning("Unexpected key %s", key)
class mqtt_base(common_base):
def __init__(self, mqtt_client, topic, default_values=None):
super().__init__(default_values)
#
self.mqtt_client = mqtt_client
self.topic = topic
for entry in self.topic.split('/'):
self.logger = self.logger.getChild(entry)
class videv_base(mqtt_base):
KEY_INFO = '__info__'
#
SET_TOPIC = "set"
def __init__(self, mqtt_client, topic, default_values=None):
super().__init__(mqtt_client, topic, default_values=default_values)
self.__display_dict__ = {}
self.__control_dict__ = {}
self.__periodic__ = task.periodic(300, self.send_all)
self.__periodic__.run()
def send_all(self, rt):
try:
for key in self:
if self[key] is not None:
self.__tx__(key, self[key])
except RuntimeError:
self.logger.warning("Runtimeerror while sending cyclic videv information. This may happen on startup.")
def add_display(self, my_key, ext_device, ext_key, on_change_only=True):
"""
listen to data changes of ext_device and update videv information
"""
if my_key not in self.keys():
self[my_key] = None
if ext_device.__class__.__name__ == "group":
# store information to identify callback from ext_device
self.__display_dict__[(id(ext_device[0]), ext_key)] = my_key
# register a callback to listen for data from external device
ext_device[0].add_callback(ext_key, None, self.__rx_ext_device_data__, on_change_only, init_now=True)
else:
# store information to identify callback from ext_device
self.__display_dict__[(id(ext_device), ext_key)] = my_key
# register a callback to listen for data from external device
ext_device.add_callback(ext_key, None, self.__rx_ext_device_data__, on_change_only, init_now=True)
# send initial display data to videv interface
data = ext_device.get(ext_key)
if data is not None:
self.__tx__(my_key, data)
def __rx_ext_device_data__(self, ext_device, ext_key, data):
my_key = self.__display_dict__[(id(ext_device), ext_key)]
self.set(my_key, data)
self.__tx__(my_key, data)
def __tx__(self, key, data):
if type(data) not in (str, ):
data = json.dumps(data)
self.mqtt_client.send('/'.join([self.topic, key]), data)
def add_control(self, my_key, ext_device, ext_key, on_change_only=True):
"""
listen to videv information and pass data to ext_device
"""
if my_key not in self.keys():
self[my_key] = None
# store information to identify callback from videv
self.__control_dict__[my_key] = (ext_device, ext_key, on_change_only)
# add callback for videv changes
self.mqtt_client.add_callback('/'.join([self.topic, my_key, self.SET_TOPIC]), self.__rx_videv_data__)
def __rx_videv_data__(self, client, userdata, message):
my_key = message.topic.split('/')[-2]
try:
data = json.loads(message.payload)
except json.decoder.JSONDecodeError:
data = message.payload
ext_device, ext_key, on_change_only = self.__control_dict__[my_key]
if my_key in self.keys():
if data != self[my_key] or not on_change_only:
ext_device.send_command(ext_key, data)
else:
self.logger.info("Ignoring rx message with topic %s", message.topic)
def add_routing(self, my_key, ext_device, ext_key, on_change_only_disp=True, on_change_only_videv=True):
"""
listen to data changes of ext_device and update videv information
and
listen to videv information and pass data to ext_device
"""
# add display
self.add_display(my_key, ext_device, ext_key, on_change_only_disp)
self.add_control(my_key, ext_device, ext_key, on_change_only_videv)

14
config.py Normal file
View File

@ -0,0 +1,14 @@
import geo
import logging
DEBUG = False # False: logging to stdout with given LOGLEVEL - True: logging all to localhost:19996 and warnings or higher to stdout
LOGLEVEL = logging.DEBUG # INFO
GEO_POSITION = geo.gps.coordinate(lat=49.519167, lon=9.3672222)
APP_NAME = "smart_enlife"
MQTT_SERVER = "mqtt.home"
MQTT_PORT = 1883
MQTT_USER = "smarthome"
MQTT_PASSWORD = ""

103
devices/__init__.py Normal file
View File

@ -0,0 +1,103 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
"""
devices (DEVICES)
=================
**Author:**
* Dirk Alders <sudo-dirk@mount-mockery.de>
**Description:**
This Module supports smarthome devices
**Submodules:**
* :mod:`shelly`
* :mod:`silvercrest_powerplug`
**Unittest:**
See also the :download:`unittest <devices/_testresults_/unittest.pdf>` documentation.
**Module Documentation:**
"""
try:
from config import APP_NAME as ROOT_LOGGER_NAME
except ImportError:
ROOT_LOGGER_NAME = 'root'
from devices.shelly import shelly as shelly_sw1
from devices.tradfri import tradfri_light as tradfri_sw
from devices.tradfri import tradfri_light as tradfri_sw_br
from devices.tradfri import tradfri_light as tradfri_sw_br_ct
from devices.tradfri import tradfri_button as tradfri_button
from devices.tradfri import tradfri_light as livarno_sw_br_ct
from devices.brennenstuhl import brennenstuhl_heatingvalve
from devices.silvercrest import silvercrest_powerplug
from devices.silvercrest import silvercrest_motion_sensor
from devices.mydevices import powerplug as my_powerplug
from devices.mydevices import audio_status
from devices.mydevices import remote
my_ambient = None
class group(object):
def __init__(self, *args):
super().__init__()
self._members = args
self._iter_counter = 0
#
self.methods = []
self.variables = []
for name in [m for m in args[0].__class__.__dict__.keys()]:
if not name.startswith('_') and callable(getattr(args[0], name)): # add all public callable attributes to the list
self.methods.append(name)
if not name.startswith('_') and not callable(getattr(args[0], name)): # add all public callable attributes to the list
self.variables.append(name)
#
for member in self:
methods = [m for m in member.__class__.__dict__.keys() if not m.startswith(
'_') if not m.startswith('_') and callable(getattr(args[0], m))]
if self.methods != methods:
raise ValueError("All given instances needs to have same methods:", self.methods, methods)
#
variables = [v for v in member.__class__.__dict__.keys() if not v.startswith(
'_') if not v.startswith('_') and not callable(getattr(args[0], v))]
if self.variables != variables:
raise ValueError("All given instances needs to have same variables:", self.variables, variables)
def __iter__(self):
return self
def __next__(self):
if self._iter_counter < len(self):
self._iter_counter += 1
return self._members[self._iter_counter - 1]
self._iter_counter = 0
raise StopIteration
def __getitem__(self, i):
return self._members[i]
def __len__(self):
return len(self._members)
def __getattribute__(self, name):
def group_execution(*args, **kwargs):
for member in self[:]:
m = getattr(member, name)
m(*args, **kwargs)
try:
rv = super().__getattribute__(name)
except AttributeError:
if callable(getattr(self[0], name)):
return group_execution
else:
return getattr(self[0], name)
else:
return rv

134
devices/base.py Normal file
View File

@ -0,0 +1,134 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
from base import mqtt_base
from base import videv_base
import json
import time
def is_json(data):
try:
json.loads(data)
except json.decoder.JSONDecodeError:
return False
else:
return True
class base(mqtt_base):
TX_TOPIC = "set"
TX_VALUE = 0
TX_DICT = 1
TX_TYPE = -1
TX_FILTER_DATA_KEYS = []
#
RX_KEYS = []
RX_IGNORE_TOPICS = []
RX_IGNORE_KEYS = []
RX_FILTER_DATA_KEYS = []
#
CFG_DATA = {}
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic, default_values=dict.fromkeys(self.RX_KEYS))
# data storage
self.__cfg_by_mid__ = None
# initialisations
mqtt_client.add_callback(topic=self.topic, callback=self.receive_callback)
mqtt_client.add_callback(topic=self.topic+"/#", callback=self.receive_callback)
#
self.add_callback(None, None, self.__state_logging__, on_change_only=True)
def __cfg_callback__(self, key, value, mid):
if self.CFG_DATA.get(key) != value and self.__cfg_by_mid__ != mid and mid is not None:
self.__cfg_by_mid__ = mid
self.logger.warning("Differing configuration identified: Sending default configuration to defice: %s", repr(self.CFG_DATA))
if self.TX_TYPE == self.TX_DICT:
self.mqtt_client.send(self.topic + '/' + self.TX_TOPIC, json.dumps(self.CFG_DATA))
else:
for key in self.CFG_DATA:
self.send_command(key, self.CFG_DATA.get(key))
def set(self, key, data, mid=None, block_callback=[]):
if key in self.CFG_DATA:
self.__cfg_callback__(key, data, mid)
if key in self.RX_IGNORE_KEYS:
pass # ignore these keys
elif key in self.RX_KEYS:
return super().set(key, data, block_callback)
else:
self.logger.warning("Unexpected key %s", key)
def receive_callback(self, client, userdata, message):
if message.topic != self.topic + '/' + videv_base.KEY_INFO:
content_key = message.topic[len(self.topic) + 1:]
if content_key not in self.RX_IGNORE_TOPICS and (not message.topic.endswith(self.TX_TOPIC) or len(self.TX_TOPIC) == 0):
self.logger.debug("Unpacking content_key \"%s\" from message.", content_key)
if is_json(message.payload):
data = json.loads(message.payload)
if type(data) is dict:
for key in data:
self.set(key, self.__device_to_instance_filter__(key, data[key]), message.mid)
else:
self.set(content_key, self.__device_to_instance_filter__(content_key, data), message.mid)
# String
else:
self.set(content_key, self.__device_to_instance_filter__(content_key, message.payload.decode('utf-8')), message.mid)
else:
self.logger.debug("Ignoring topic %s", content_key)
def __device_to_instance_filter__(self, key, data):
if key in self.RX_FILTER_DATA_KEYS:
if data in [1, 'on', 'ON']:
return True
elif data in [0, 'off', 'OFF']:
return False
return data
def __instance_to_device_filter__(self, key, data):
if key in self.TX_FILTER_DATA_KEYS:
if data is True:
return "on"
elif data is False:
return "off"
return data
def send_command(self, key, data):
data = self.__instance_to_device_filter__(key, data)
if self.TX_TOPIC is not None:
if self.TX_TYPE < 0:
self.logger.error("Unknown tx type. Set TX_TYPE of class to a known value")
else:
self.logger.debug("Sending data for %s - %s", key, str(data))
if self.TX_TYPE == self.TX_DICT:
try:
self.mqtt_client.send('/'.join([self.topic, self.TX_TOPIC]), json.dumps({key: data}))
except TypeError:
print(self.topic)
print(key.__dict__)
print(key)
print(data)
raise TypeError
else:
if type(data) not in [str, bytes]:
data = json.dumps(data)
self.mqtt_client.send('/'.join([self.topic, key, self.TX_TOPIC] if len(self.TX_TOPIC) > 0 else [self.topic, key]), data)
else:
self.logger.error("Unknown tx toptic. Set TX_TOPIC of class to a known value")
class base_output(base):
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
self.__all_off_enabled__ = True
def disable_all_off(self, state=True):
self.__all_off_enabled__ = not state
def all_off(self):
if self.__all_off_enabled__:
try:
self.__all_off__()
except (AttributeError, TypeError) as e:
self.logger.warning("Method all_off was used, but __all_off__ method wasn't callable: %s", repr(e))

119
devices/brennenstuhl.py Normal file
View File

@ -0,0 +1,119 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
from devices.base import base
import json
import task
import time
class brennenstuhl_heatingvalve(base):
""" Communication (MQTT)
brennenstuhl_heatingvalve {
| "away_mode": ["ON", "OFF"]
| "battery": [0...100] %
| "child_lock": ["LOCK", "UNLOCK"]
| "current_heating_setpoint": [5...30] °C
| "linkquality": [0...255] lqi
| "local_temperature": [numeric] °C
| "preset": ["manual", ...]
| "system_mode": ["heat", ...]
| "valve_detection": ["ON", "OFF"]
| "window_detection": ["ON", "OFF"]
| }
+- set {
"away_mode": ["ON", "OFF", "TOGGLE"]
"child_lock": ["LOCK", "UNLOCK"]
"current_heating_setpoint": [5...30] °C
"preset": ["manual", ...]
"system_mode": ["heat", ...]
"valve_detection": ["ON", "OFF", "TOGGLE"]
"window_detection": ["ON", "OFF", "TOGGLE"]
}
"""
KEY_LINKQUALITY = "linkquality"
KEY_BATTERY = "battery"
KEY_HEATING_SETPOINT = "current_heating_setpoint"
KEY_TEMPERATURE = "local_temperature"
#
KEY_AWAY_MODE = "away_mode"
KEY_CHILD_LOCK = "child_lock"
KEY_PRESET = "preset"
KEY_SYSTEM_MODE = "system_mode"
KEY_VALVE_DETECTION = "valve_detection"
KEY_WINDOW_DETECTION = "window_detection"
#
RETRY_CYCLE_TIME = 2.5
MAX_TX_RETRIES = 20
RETRY_TIMEOUT = RETRY_CYCLE_TIME * MAX_TX_RETRIES
#
TX_TYPE = base.TX_DICT
#
RX_KEYS = [KEY_LINKQUALITY, KEY_BATTERY, KEY_HEATING_SETPOINT, KEY_TEMPERATURE]
RX_IGNORE_KEYS = [KEY_AWAY_MODE, KEY_CHILD_LOCK, KEY_PRESET, KEY_SYSTEM_MODE, KEY_VALVE_DETECTION, KEY_WINDOW_DETECTION]
#
CFG_DATA = {
KEY_WINDOW_DETECTION: "ON",
KEY_VALVE_DETECTION: "ON",
KEY_SYSTEM_MODE: "heat",
KEY_PRESET: "manual"
}
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
self.add_callback(self.KEY_HEATING_SETPOINT, None, self.__valave_temp_rx__)
self.__tx_temperature__ = None
self.__rx_temperature__ = None
self.__tx_timestamp__ = 0
#
self.task = task.periodic(self.RETRY_CYCLE_TIME, self.__task__)
self.task.run()
def __state_logging__(self, inst, key, data):
if key in [self.KEY_HEATING_SETPOINT, self.KEY_CHILD_LOCK, self.KEY_WINDOW_DETECTION, self.KEY_VALVE_DETECTION]:
self.logger.info("State change of '%s' to '%s'", key, repr(data))
def send_command(self, key, data):
if key == self.KEY_HEATING_SETPOINT:
self.__tx_temperature__ = data
self.__tx_timestamp__ = time.time()
base.send_command(self, key, data)
def __valave_temp_rx__(self, inst, key, data):
if key == self.KEY_HEATING_SETPOINT:
self.__rx_temperature__ = data
def __task__(self, rt):
if self.__tx_temperature__ is not None and self.__tx_timestamp__ is not None: # Already send a setpoint
if self.__tx_temperature__ != self.__rx_temperature__: # Setpoint and valve feedback are unequal
if time.time() - self.__tx_timestamp__ < self.RETRY_TIMEOUT: # Timeout condition allows resend of setpoint
self.logger.warning("Setpoint not yet acknoledged by device. Sending setpoint again")
self.set_heating_setpoint(self.__tx_temperature__)
return
else:
self.__tx_timestamp__ = None # Disable resend logic, if setpoint and valve setpoint are equal
#
# RX
#
@property
def linkqulity(self):
return self.get(self.KEY_LINKQUALITY)
@property
def heating_setpoint(self):
return self.get(self.KEY_HEATING_SETPOINT)
@property
def temperature(self):
return self.get(self.KEY_TEMPERATURE)
#
# TX
#
def set_heating_setpoint(self, setpoint):
self.send_command(self.KEY_HEATING_SETPOINT, setpoint)
def set_heating_setpoint_mcb(self, device, key, data):
self.set_heating_setpoint(data)

262
devices/mydevices.py Normal file
View File

@ -0,0 +1,262 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
from devices.base import base, base_output
import logging
class powerplug(base_output):
""" Communication (MQTT)
my_powerplug
+- output
+- 1 [True, False] <- status
| +- set [True, False, "toggle"] <- command
+- 2 [True, False] <- status
| +- set [True, False, "toggle"] <- command
+- 3 [True, False] <- status
| +- set [True, False, "toggle"] <- command
+- 4 [True, False] <- status
| +- set [True, False, "toggle"] <- command
+- all
+- set [True, False, "toggle"] <- command
"""
KEY_OUTPUT_0 = "output/1"
KEY_OUTPUT_1 = "output/2"
KEY_OUTPUT_2 = "output/3"
KEY_OUTPUT_3 = "output/4"
KEY_OUTPUT_ALL = "output/all"
KEY_OUTPUT_LIST = [KEY_OUTPUT_0, KEY_OUTPUT_1, KEY_OUTPUT_2, KEY_OUTPUT_3]
#
TX_TYPE = base.TX_VALUE
#
RX_KEYS = [KEY_OUTPUT_0, KEY_OUTPUT_1, KEY_OUTPUT_2, KEY_OUTPUT_3]
def __state_logging__(self, inst, key, data):
if key in self.KEY_OUTPUT_LIST:
self.logger.info("State change of '%s' to '%s'", key, repr(data))
#
# RX
#
@property
def output_0(self):
"""rv: [True, False]"""
return self.get(self.KEY_OUTPUT_0)
@property
def output_1(self):
"""rv: [True, False]"""
return self.get(self.KEY_OUTPUT_1)
@property
def output_2(self):
"""rv: [True, False]"""
return self.get(self.KEY_OUTPUT_2)
@property
def output_3(self):
"""rv: [True, False]"""
return self.get(self.KEY_OUTPUT_3)
#
# TX
#
def set_output(self, key, state):
if key in self.KEY_OUTPUT_LIST:
self.send_command(key, state)
else:
logging.error("Unknown key to set the output!")
def set_output_0(self, state):
"""state: [True, False]"""
self.send_command(self.KEY_OUTPUT_0, state)
def set_output_0_mcb(self, device, key, data):
self.set_output_0(data)
def toggle_output_0_mcb(self, device, key, data):
self.set_output_0(not self.output_0)
def set_output_1(self, state):
"""state: [True, False]"""
self.send_command(self.KEY_OUTPUT_1, state)
def set_output_1_mcb(self, device, key, data):
self.set_output_1(data)
def toggle_output_1_mcb(self, device, key, data):
self.set_output_1(not self.output_1)
def set_output_2(self, state):
"""state: [True, False]"""
self.send_command(self.KEY_OUTPUT_2, state)
def set_output_2_mcb(self, device, key, data):
self.set_output_2(data)
def toggle_output_2_mcb(self, device, key, data):
self.set_output_2(not self.output_2)
def set_output_3(self, state):
"""state: [True, False]"""
self.send_command(self.KEY_OUTPUT_3, state)
def set_output_3_mcb(self, device, key, data):
self.set_output_3(data)
def toggle_output_3_mcb(self, device, key, data):
self.set_output_3(not self.output_3)
def set_output_all(self, state):
"""state: [True, False, 'toggle']"""
self.send_command(self.KEY_OUTPUT_ALL, state)
def set_output_all_mcb(self, device, key, data):
self.set_output_all(data)
def __all_off__(self):
self.set_output_all(False)
class remote(base):
""" Communication (MQTT)
remote (RAS5) <- command
+- CD [dc]
+- LINE1 [dc]
+- LINE2 [dc]
+- LINE3 [dc]
+- MUTE [dc]
+- POWER [dc]
+- VOLDOWN [dc]
+- VOLUP [dc]
+- PHONO [dc]
+- DOCK [dc]
remote (EUR642100) <- command
+- OPEN_CLOSE [dc]
+- VOLDOWN [dc]
+- VOLUP [dc]
+- ONE [dc]
+- TWO [dc]
+- THREE [dc]
+- FOUR [dc]
+- FIVE [dc]
+- SIX [dc]
+- SEVEN [dc]
+- EIGHT [dc]
+- NINE [dc]
+- ZERO [dc]
+- TEN [dc]
+- TEN_PLUS [dc]
+- PROGRAM [dc]
+- CLEAR [dc]
+- RECALL [dc]
+- TIME_MODE [dc]
+- A_B_REPEAT [dc]
+- REPEAT [dc]
+- RANDOM [dc]
+- AUTO_CUE [dc]
+- TAPE_LENGTH [dc]
+- SIDE_A_B [dc]
+- TIME_FADE [dc]
+- PEAK_SEARCH [dc]
+- SEARCH_BACK [dc]
+- SEARCH_FOR [dc]
+- TRACK_NEXT [dc]
+- TRACK_PREV [dc]
+- STOP [dc]
+- PAUSE [dc]
+- PLAY [dc]
"""
KEY_CD = "CD"
KEY_LINE1 = "LINE1"
KEY_LINE2 = "LINE2"
KEY_LINE3 = "LINE3"
KEY_PHONO = "PHONO"
KEY_MUTE = "MUTE"
KEY_POWER = "POWER"
KEY_VOLDOWN = "VOLDOWN"
KEY_VOLUP = "VOLUP"
#
TX_TOPIC = ''
TX_TYPE = base.TX_VALUE
#
RX_IGNORE_TOPICS = [KEY_CD, KEY_LINE1, KEY_LINE2, KEY_LINE3, KEY_PHONO, KEY_MUTE, KEY_POWER, KEY_VOLUP, KEY_VOLDOWN]
def __state_logging__(self, inst, key, data):
pass # This is just a TX device using self.set_*
def set_cd(self, device=None, key=None, data=None):
self.logger.info("Changing amplifier source to CD")
self.send_command(self.KEY_CD, None)
def set_line1(self, device=None, key=None, data=None):
self.logger.info("Changing amplifier source to LINE1")
self.send_command(self.KEY_LINE1, None)
def set_line2(self, device=None, key=None, data=None):
self.logger.info("Changing amplifier source to LINE2")
self.send_command(self.KEY_LINE2, None)
def set_line3(self, device=None, key=None, data=None):
self.logger.info("Changing amplifier source to LINE3")
self.send_command(self.KEY_LINE3, None)
def set_phono(self, device=None, key=None, data=None):
self.logger.info("Changing amplifier source to PHONO")
self.send_command(self.KEY_PHONO, None)
def set_mute(self, device=None, key=None, data=None):
self.logger.info("Muting / Unmuting amplifier")
self.send_command(self.KEY_MUTE, None)
def set_power(self, device=None, key=None, data=None):
self.logger.info("Power on/off amplifier")
self.send_command(self.KEY_POWER, None)
def set_volume_up(self, data=False):
"""data: [True, False]"""
self.logger.info("Increasing amplifier volume")
self.send_command(self.KEY_VOLUP, data)
def set_volume_down(self, data=False):
"""data: [True, False]"""
self.logger.info("Decreasing amplifier volume")
self.send_command(self.KEY_VOLDOWN, data)
def default_inc(self, device=None, key=None, data=None):
self.set_volume_up(True)
def default_dec(self, device=None, key=None, data=None):
self.set_volume_down(True)
def default_stop(self, device=None, key=None, data=None):
self.set_volume_up(False)
class audio_status(base):
""" Communication (MQTT)
audio_status
+- state [True, False] <- status
+- title [text] <- status
"""
KEY_STATE = "state"
KEY_TITLE = "title"
#
TX_TYPE = base.TX_VALUE
#
RX_KEYS = [KEY_STATE, KEY_TITLE]
def __state_logging__(self, inst, key, data):
if key in [self.KEY_STATE, self.KEY_TITLE]:
self.logger.info("State change of '%s' to '%s'", key, repr(data))
def set_state(self, num, data):
"""data: [True, False]"""
self.send_command(self.KEY_STATE + "/" + str(num), data)
def set_state_mcb(self, device, key, data):
self.set_state(data)

171
devices/shelly.py Normal file
View File

@ -0,0 +1,171 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
from devices.base import base_output
import logging
import task
class shelly(base_output):
""" Communication (MQTT)
shelly
+- relay
| +- 0 ["on" / "off"] <- status
| | +- command ["on"/ "off"] <- command
| | +- energy [numeric] <- status
| +- 1 ["on" / "off"] <- status
| +- command ["on"/ "off"] <- command
| +- energy [numeric] <- status
+- input
| +- 0 [0 / 1] <- status
| +- 1 [0 / 1] <- status
+- input_event
| +- 0 <- status
| +- 1 <- status
+- logpush
| +- 0 [0 / 1] <- status
| +- 1 [0 / 1] <- status
+- temperature [numeric] °C <- status
+- temperature_f [numeric] F <- status
+- overtemperature [0 / 1] <- status
+- id <- status
+- model <- status
+- mac <- status
+- ip <- status
+- new_fw <- status
+- fw_ver <- status
"""
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"
KEY_TEMPERATURE = "temperature"
KEY_OVERTEMPERATURE = "overtemperature"
KEY_ID = "id"
KEY_MODEL = "model"
KEY_MAC = "mac"
KEY_IP = "ip"
KEY_NEW_FIRMWARE = "new_fw"
KEY_FIRMWARE_VERSION = "fw_ver"
#
TX_TOPIC = "command"
TX_TYPE = base_output.TX_VALUE
TX_FILTER_DATA_KEYS = [KEY_OUTPUT_0, KEY_OUTPUT_1]
#
RX_KEYS = [KEY_OUTPUT_0, KEY_OUTPUT_1, KEY_INPUT_0, KEY_INPUT_1, KEY_LONGPUSH_0, KEY_LONGPUSH_1, KEY_OVERTEMPERATURE, KEY_TEMPERATURE,
KEY_ID, KEY_MODEL, KEY_MAC, KEY_IP, KEY_NEW_FIRMWARE, KEY_FIRMWARE_VERSION]
RX_IGNORE_TOPICS = [KEY_OUTPUT_0 + '/' + "energy", KEY_OUTPUT_1 + '/' + "energy", 'input_event/0', 'input_event/1']
RX_IGNORE_KEYS = ['temperature_f']
RX_FILTER_DATA_KEYS = [KEY_INPUT_0, KEY_INPUT_1, KEY_LONGPUSH_0, KEY_LONGPUSH_1, KEY_OUTPUT_0, KEY_OUTPUT_1, KEY_OVERTEMPERATURE]
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
#
self.output_key_delayed = None
self.delayed_flash_task = task.delayed(0.3, self.flash_task)
self.delayed_off_task = task.delayed(0.3, self.off_task)
#
self.all_off_requested = False
def __state_logging__(self, inst, key, data):
if key in [self.KEY_OUTPUT_0, self.KEY_OUTPUT_1]:
self.logger.info("State change of '%s' to '%s'", key, repr(data))
elif key in [self.KEY_INPUT_0, self.KEY_INPUT_1, self.KEY_LONGPUSH_0, self.KEY_LONGPUSH_1]:
self.logger.info("Input action '%s' with '%s'", key, repr(data))
def flash_task(self, *args):
if self.flash_active:
self.send_command(self.output_key_delayed, not self.get(self.output_key_delayed))
self.output_key_delayed = None
if self.all_off_requested:
self.delayed_off_task.run()
def off_task(self, *args):
self.all_off()
@property
def flash_active(self):
return self.output_key_delayed is not None
#
# RX
#
@property
def output_0(self):
"""rv: [True, False]"""
return self.get(self.KEY_OUTPUT_0)
@property
def output_1(self):
"""rv: [True, False]"""
return self.get(self.KEY_OUTPUT_1)
@property
def input_0(self):
"""rv: [True, False]"""
return self.get(self.KEY_INPUT_0)
@property
def input_1(self):
"""rv: [True, False]"""
return self.get(self.KEY_INPUT_1)
@property
def longpush_0(self):
"""rv: [True, False]"""
return self.get(self.KEY_LONGPUSH_0)
@property
def longpush_1(self):
"""rv: [True, False]"""
return self.get(self.KEY_LONGPUSH_1)
@property
def temperature(self):
"""rv: numeric value"""
return self.get(self.KEY_TEMPERATURE)
#
# TX
#
def set_output_0(self, state):
"""state: [True, False]"""
self.send_command(self.KEY_OUTPUT_0, state)
def set_output_0_mcb(self, device, key, data):
self.set_output_0(data)
def toggle_output_0_mcb(self, device, key, data):
self.set_output_0(not self.output_0)
def set_output_1(self, state):
"""state: [True, False]"""
self.send_command(self.KEY_OUTPUT_1, state)
def set_output_1_mcb(self, device, key, data):
self.set_output_1(data)
def toggle_output_1_mcb(self, device, key, data):
self.set_output_1(not self.output_1)
def flash_0_mcb(self, device, key, data):
self.output_key_delayed = self.KEY_OUTPUT_0
self.toggle_output_0_mcb(device, key, data)
self.delayed_flash_task.run()
def flash_1_mcb(self, device, key, data):
self.output_key_delayed = self.KEY_OUTPUT_1
self.toggle_output_1_mcb(device, key, data)
self.delayed_flash_task.run()
def __all_off__(self):
if self.flash_active:
self.all_off_requested = True
else:
if self.output_0:
self.set_output_0(False)
if self.output_1:
self.set_output_1(False)

107
devices/silvercrest.py Normal file
View File

@ -0,0 +1,107 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
from devices.base import base, base_output
import logging
class silvercrest_powerplug(base_output):
""" Communication (MQTT)
silvercrest_powerplug {
| "state": ["ON" / "OFF"]
| "linkquality": [0...255] lqi
| }
+- get {
| "state": ""
| }
+- set {
"state": ["ON" / "OFF"]
}
"""
KEY_LINKQUALITY = "linkquality"
KEY_OUTPUT_0 = "state"
#
TX_TYPE = base.TX_DICT
TX_FILTER_DATA_KEYS = [KEY_OUTPUT_0]
#
RX_KEYS = [KEY_LINKQUALITY, KEY_OUTPUT_0]
RX_FILTER_DATA_KEYS = [KEY_OUTPUT_0]
def __state_logging__(self, inst, key, data):
if key in [self.KEY_OUTPUT_0]:
self.logger.info("State change of '%s' to '%s'", key, repr(data))
#
# RX
#
@property
def output_0(self):
"""rv: [True, False]"""
return self.get(self.KEY_OUTPUT_0)
@property
def linkquality(self):
"""rv: numeric value"""
return self.get(self.KEY_LINKQUALITY)
#
# TX
#
def set_output_0(self, state):
"""state: [True, False]"""
self.send_command(self.KEY_OUTPUT_0, state)
def set_output_0_mcb(self, device, key, data):
self.set_output_0(data)
def toggle_output_0_mcb(self, device, key, data):
self.set_output_0(not self.output_0)
def __all_off__(self):
if self.output_0:
self.set_output_0(False)
class silvercrest_motion_sensor(base):
""" Communication (MQTT)
silvercrest_motion_sensor {
battery: [0...100] %
battery_low: [True, False]
linkquality: [0...255] lqi
occupancy: [True, False]
tamper: [True, False]
voltage: [0...] mV
}
"""
KEY_BATTERY = "battery"
KEY_BATTERY_LOW = "battery_low"
KEY_LINKQUALITY = "linkquality"
KEY_OCCUPANCY = "occupancy"
KEY_UNMOUNTED = "tamper"
KEY_VOLTAGE = "voltage"
#
TX_TYPE = base.TX_DICT
#
RX_KEYS = [KEY_BATTERY, KEY_BATTERY_LOW, KEY_LINKQUALITY, KEY_OCCUPANCY, KEY_UNMOUNTED, KEY_VOLTAGE]
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
def __state_logging__(self, inst, key, data):
if key in [self.KEY_OCCUPANCY, self.KEY_UNMOUNTED]:
self.logger.info("State change of '%s' to '%s'", key, repr(data))
#
# RX
#
@property
def linkquality(self):
"""rv: numeric value"""
return self.get(self.KEY_LINKQUALITY)
@property
def battery(self):
"""rv: numeric value"""
return self.get(self.KEY_BATTERY)

192
devices/tradfri.py Normal file
View File

@ -0,0 +1,192 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
from devices.base import base, base_output
import logging
class tradfri_light(base_output):
""" Communication (MQTT)
tradfri_light {
| "state": ["ON" / "OFF" / "TOGGLE"]
| "linkquality": [0...255] lqi
| "brightness": [0...254]
| "color_mode": ["color_temp"]
| "color_temp": ["coolest", "cool", "neutral", "warm", "warmest", 250...454]
| "color_temp_startup": ["coolest", "cool", "neutral", "warm", "warmest", "previous", 250...454]
| "update": []
| }
+- get {
| "state": ""
| }
+- set {
"state": ["ON" / "OFF"]
"brightness": [0...256]
"color_temp": [250...454]
"transition": [0...] seconds
"brightness_move": [-X...0...X] X/s
"brightness_step": [-X...0...X]
"color_temp_move": [-X...0...X] X/s
"color_temp_step": [-X...0...X]
}
"""
KEY_LINKQUALITY = "linkquality"
KEY_OUTPUT_0 = "state"
KEY_BRIGHTNESS = "brightness"
KEY_COLOR_TEMP = "color_temp"
KEY_BRIGHTNESS_FADE = "brightness_move"
#
TX_TYPE = base.TX_DICT
TX_FILTER_DATA_KEYS = [KEY_OUTPUT_0, KEY_BRIGHTNESS, KEY_COLOR_TEMP, KEY_BRIGHTNESS_FADE]
#
RX_KEYS = [KEY_LINKQUALITY, KEY_OUTPUT_0, KEY_BRIGHTNESS, KEY_COLOR_TEMP]
RX_IGNORE_KEYS = ['update', 'color_mode', 'color_temp_startup']
RX_FILTER_DATA_KEYS = [KEY_OUTPUT_0, KEY_BRIGHTNESS, KEY_COLOR_TEMP]
def __state_logging__(self, inst, key, data):
if key in [self.KEY_OUTPUT_0, self.KEY_BRIGHTNESS, self.KEY_COLOR_TEMP, self.KEY_BRIGHTNESS_FADE]:
self.logger.info("State change of '%s' to '%s'", key, repr(data))
def __device_to_instance_filter__(self, key, data):
if key == self.KEY_BRIGHTNESS:
return int(round((data - 1) * 100 / 253, 0))
elif key == self.KEY_COLOR_TEMP:
return int(round((data - 250) * 10 / 204, 0))
return super().__device_to_instance_filter__(key, data)
def __instance_to_device_filter__(self, key, data):
if key == self.KEY_BRIGHTNESS:
return int(round(data * 253 / 100 + 1, 0))
elif key == self.KEY_COLOR_TEMP:
return int(round(data * 204 / 10 + 250, 0))
return super().__instance_to_device_filter__(key, data)
#
# RX
#
@property
def output_0(self):
"""rv: [True, False]"""
return self.get(self.KEY_OUTPUT_0, False)
@property
def linkquality(self):
"""rv: numeric value"""
return self.get(self.KEY_LINKQUALITY, 0)
@property
def brightness(self):
"""rv: numeric value [0%, ..., 100%]"""
return self.get(self.KEY_BRIGHTNESS, 0)
@property
def color_temp(self):
"""rv: numeric value [0, ..., 10]"""
return self.get(self.KEY_COLOR_TEMP, 0)
#
# TX
#
def request_data(self, device=None, key=None, data=None):
self.mqtt_client.send(self.topic + "/get", '{"%s": ""}' % self.KEY_OUTPUT_0)
def set_output_0(self, state):
"""state: [True, False]"""
self.send_command(self.KEY_OUTPUT_0, state)
def set_output_0_mcb(self, device, key, data):
self.set_output_0(data)
def toggle_output_0_mcb(self, device, key, data):
self.set_output_0(not self.output_0)
def set_brightness(self, brightness):
"""brightness: [0, ..., 100]"""
self.send_command(self.KEY_BRIGHTNESS, brightness)
def set_brightness_mcb(self, device, key, data):
self.set_brightness(data)
def default_inc(self, speed=40):
self.send_command(self.KEY_BRIGHTNESS_FADE, speed)
def default_dec(self, speed=-40):
self.default_inc(speed)
def default_stop(self):
self.default_inc(0)
def set_color_temp(self, color_temp):
"""color_temp: [0, ..., 10]"""
self.send_command(self.KEY_COLOR_TEMP, color_temp)
def set_color_temp_mcb(self, device, key, data):
self.set_color_temp(data)
def __all_off__(self):
if self.output_0:
self.set_output_0(False)
class tradfri_button(base):
""" Communication (MQTT)
tradfri_button {
"action": [
"arrow_left_click",
"arrow_left_hold",
"arrow_left_release",
"arrow_right_click",
"arrow_right_hold",
"arrow_right_release",
"brightness_down_click",
"brightness_down_hold",
"brightness_down_release",
"brightness_up_click",
"brightness_up_hold",
"brightness_up_release",
"toggle"
]
"action_duration": [0...] s
"battery": [0...100] %
"linkquality": [0...255] lqi
"update": []
}
"""
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_UP_RELEASE = "brightness_up_release"
ACTION_BRIGHTNESS_DOWN_LONG = "brightness_down_hold"
ACTION_BRIGHTNESS_DOWN_RELEASE = "brightness_down_release"
ACTION_RIGHT_LONG = "arrow_right_hold"
ACTION_RIGHT_RELEASE = "arrow_right_release"
ACTION_LEFT_LONG = "arrow_left_hold"
ACTION_LEFT_RELEASE = "arrow_left_release"
#
KEY_LINKQUALITY = "linkquality"
KEY_BATTERY = "battery"
KEY_ACTION = "action"
KEY_ACTION_DURATION = "action_duration"
#
RX_KEYS = [KEY_LINKQUALITY, KEY_BATTERY, KEY_ACTION]
RX_IGNORE_KEYS = ['update', KEY_ACTION_DURATION]
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
def __state_logging__(self, inst, key, data):
if key in [self.KEY_ACTION]:
self.logger.info("Input '%s' with '%s'", key, repr(data))
#
# RX
#
@property
def action(self):
"""rv: action_txt"""
return self.get(self.KEY_ACTION)

1
geo Submodule

@ -0,0 +1 @@
Subproject commit 11166bb27ad2335f7812fcb88c788397f5106751

1
mqtt Submodule

@ -0,0 +1 @@
Subproject commit 14e56ccdbf6594f699b4afcfb4acafe9b899e914

1
report Submodule

@ -0,0 +1 @@
Subproject commit 7003c13ef8c7e7c3a55a545cbbad4039cc024a9f

129
smart_enlife.py Normal file
View File

@ -0,0 +1,129 @@
import config
import devices
import geo
import logging
import geo.sun
import mqtt
import random
import report
import state_machine
import time
logger = logging.getLogger(config.APP_NAME)
class day_states(state_machine.state_machine):
LOG_PREFIX = 'Day states:'
STATE_NIGHT = 'state_night'
STATE_MORNING = 'state_morning'
STATE_DAY = 'state_day'
STATE_EVENING = 'state_evening'
CONDITION_NIGHT_END = 'condition_night_end'
CONDITION_MORNING_END = 'condition_morning_end'
CONDITION_DAY_END = 'condition_day_end'
CONDITION_EVENING_END = 'condition_evening_end'
TRANSITIONS = {
STATE_NIGHT: (
(CONDITION_NIGHT_END, 1, STATE_MORNING),
),
STATE_MORNING: (
(CONDITION_MORNING_END, 1, STATE_DAY),
),
STATE_DAY: (
(CONDITION_DAY_END, 1, STATE_EVENING),
),
STATE_EVENING: (
(CONDITION_EVENING_END, 1, STATE_NIGHT),
),
}
def condition_night_end(self):
ltime = time.localtime(time.time())
return ltime.tm_hour >= 5 and ltime.tm_min >= 20 and not self.condition_morning_end() # 2nd condition to ensure day change
def condition_morning_end(self):
ltime = time.mktime(time.localtime(time.time()))
sunrise = time.mktime(geo.sun.sunrise(config.GEO_POSITION)) + 30 * 60
return ltime > sunrise
def condition_day_end(self):
ltime = time.mktime(time.localtime(time.time()))
sunset = time.mktime(geo.sun.sunset(config.GEO_POSITION)) - 30 * 60
return ltime > sunset
def condition_evening_end(self):
ltime = time.localtime(time.time())
return ltime.tm_hour >= 20 and ltime.tm_min >= 30
def get_sorted_sw_offs(num):
SWITCH_DURATION = 30 * 60
rv = []
for i in range(0, num):
rv.append(random.randint(0, SWITCH_DURATION))
rv.sort()
return rv
def switch_x(state: bool, sm: day_states, mydevs: list):
tm = time.time()
random.shuffle(mydevs)
offsets = get_sorted_sw_offs(len(mydevs))
logger.info("State changed to %s with offsets = %s", repr(sm.this_state()), repr(offsets))
allowed_states = [sm.STATE_EVENING, sm.STATE_MORNING] if state else [sm.STATE_DAY, sm.STATE_NIGHT]
while (len(mydevs) > 0 and sm.this_state() in allowed_states):
sm.work()
dt = time.time() - tm
if dt > offsets[0]:
offsets.pop(0)
d: devices.tradfri_sw_br_ct = mydevs.pop()
logger.info("Swiching %s to state %s", d.topic, repr(state))
d.set_output_0(state)
time.sleep(0.25)
def switch_on(sm: day_states, devs: list):
switch_x(True, sm, devs[:])
def switch_off(sm: day_states, devs: list):
switch_x(False, sm, devs[:])
if __name__ == "__main__":
#
# Logging
#
if config.DEBUG:
report.appLoggingConfigure(None, 'stdout', ((config.APP_NAME, logging.DEBUG), ),
target_level=logging.WARNING, fmt=report.SHORT_FMT, host='localhost', port=19996)
else:
report.stdoutLoggingConfigure(((config.APP_NAME, config.LOGLEVEL), ), report.SHORT_FMT)
#
# MQTT Client
#
mc = mqtt.mqtt_client(host=config.MQTT_SERVER, port=config.MQTT_PORT, username=config.MQTT_USER,
password=config.MQTT_PASSWORD, name=config.APP_NAME)
#
# Smarthome physical Devices
devs = []
devs.append(devices.tradfri_sw_br_ct(mc, "zigbee_gfe/gfe/room1/window_light"))
devs.append(devices.tradfri_sw_br_ct(mc, "zigbee_gfe/gfe/room2/window_light"))
devs.append(devices.tradfri_sw_br_ct(mc, "zigbee_gfe/gfe/room3/window_light"))
#
# Functionality
#
sm = day_states(day_states.STATE_DAY, logging.DEBUG)
sm.register_state_change_callback(sm.STATE_NIGHT, None, switch_off, sm, devs)
sm.register_state_change_callback(sm.STATE_MORNING, None, switch_on, sm, devs)
sm.register_state_change_callback(sm.STATE_DAY, None, switch_off, sm, devs)
sm.register_state_change_callback(sm.STATE_EVENING, None, switch_on, sm, devs)
while True:
sm.work()
time.sleep(0.25)

4
smart_enlife.sh Executable file
View File

@ -0,0 +1,4 @@
#!/bin/sh
#
BASEPATH=`dirname $0`
$BASEPATH/venv/bin/python $BASEPATH/smart_enlife.py

1
state_machine Submodule

@ -0,0 +1 @@
Subproject commit aef99580e1cdc05e6ca80efe05461ab861d6ef39

1
task Submodule

@ -0,0 +1 @@
Subproject commit af35e83d1f07fd4cb9070bdb77cf1f3bdda3a463