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

Integration of module devdi

Browse Source
This commit is contained in:
Dirk Alders 2023-10-29 15:10:52 +01:00
parent 4a99c33b15
commit 0f429b2579
11 changed files with 139 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.vscode/settings.json vendored
View File

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

2
devdi

@ -1 +1 @@
Subproject commit a5a55a158050fdb978f4e9c1f7d43c8e6aa83a1c
Subproject commit 773d0a6679810b365bbd4537156c513b0f496c5a

25
devices/__init__.py
View File

@ -1,16 +1,23 @@
from devices.brennenstuhl import vlv as brennenstuhl_heatingvalve
from devices.livarno import sw as silvercrest_powerplug
from devices.livarno import sw_br_ct as livarno_sw_br_ct
from devices.my import powerplug as my_powerplug
from devices.shelly import shelly_sw1
from devices.tradfri import sw as tradfri_sw
from devices.tradfri import sw_br as tradfri_sw_br
from devices.tradfri import sw_br_ct as tradfri_sw_br_ct
tradfri_button = None
from devices.livarno import sw_br_ct as livarno_sw_br_ct
silvercrest_powerplug = None
tradfri_button = None # TODO: required, when a interface for external device stimulation is available
silvercrest_motion_sensor = None
from devices.brennenstuhl import vlv as brennenstuhl_heatingvalve
my_powerplug = None
audio_status = None
remote = None
remote = None
class group(object):
def __init__(self, *args):
self.device_group = args
self.topic = self.device_group[0].topic
def power_on_action(self, *args, **kwargs):
for gm in self.device_group:
gm.power_on_action(*args, **kwargs)

24
devices/base.py
View File

@ -1,5 +1,10 @@
import logging
try:
from config import APP_NAME as ROOT_LOGGER_NAME
except ImportError:
ROOT_LOGGER_NAME = 'root'
class base(dict):
"""A base device for all devicetypes
@ -15,11 +20,11 @@ class base(dict):
self.mqtt_client = mqtt_client
self.topic = topic
#
self.logger = logging.getLogger('devices')
self.logger = logging.getLogger(ROOT_LOGGER_NAME).getChild("devices")
for entry in self.topic.split('/'):
self.logger = self.logger.getChild(entry)
#
self.__power_on_inst__ = []
self.__power_on_inst__ = {}
def __set__(self, key, data):
if key in self.PROPERTIES:
@ -28,13 +33,16 @@ class base(dict):
else:
self.logger.warning("Ignoring unsupported property %s", key)
def power_on(self):
for i in self.__power_on_inst__:
def power_on(self, key):
for i in self.__power_on_inst__.get(key, []):
self.logger.info("Power on action for %s will be executed.", i.topic)
i.power_on_action()
def register_power_on_instance(self, inst):
if inst not in self.__power_on_inst__:
self.__power_on_inst__.append(inst)
def register_power_on_instance(self, inst, key):
if key not in self.__power_on_inst__:
self.__power_on_inst__[key] = []
if inst not in self.__power_on_inst__[key]:
self.__power_on_inst__[key].append(inst)
def power_on_action(self):
pass
pass

19
devices/brennenstuhl.py
View File

@ -28,6 +28,7 @@
from devices.base import base
import json
import task
import time
""" ANSWER of a device:
@ -78,6 +79,9 @@ class vlv(base):
self["window_detection"] = "ON"
#
self.mqtt_client.add_callback(self.topic + '/set', self.__rx_set__)
#
self.tq = task.threaded_queue()
self.tq.run()
def set_state(self, value):
self.__set__("state", "on" if value else "off")
@ -88,10 +92,15 @@ class vlv(base):
self.logger.info("Received set data: %s", repr(data))
for key in data:
self.__set__(key, data[key])
#time.sleep(1.5)
self.send_device_status()
self.tq.enqueue(1, self.send_device_status, data)
def send_device_status(self):
data = json.dumps(self)
def send_device_status(self, rt, data):
for i in range(0, 75):
time.sleep(0.01)
if self.tq.qsize() >= 3:
return
for key in self:
if key not in data:
data[key] = self[key]
self.logger.info("Sending status: %s", repr(data))
self.mqtt_client.send(self.topic, data)
self.mqtt_client.send(self.topic, json.dumps(data))

11
devices/livarno.py
View File

@ -1,9 +1,14 @@
import devices.tradfri
from devices.tradfri import sw as tradfri_sw
from devices.tradfri import sw_br_ct as tradfri_sw_br_ct
class sw_br_ct(devices.tradfri.sw_br_ct):
class sw(tradfri_sw):
pass
class sw_br_ct(tradfri_sw_br_ct):
def set_state(self, value):
self.__set__("state", "on" if value else "off")
def power_on_action(self):
pass
self["state"] = "on"

36
devices/my.py Normal file
View File

@ -0,0 +1,36 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
from devices.base import base
class powerplug(base):
PROPERTIES = [
"output/1",
"output/2",
"output/3",
"output/4",
]
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
#
for i in range(0, 4):
self[self.PROPERTIES[i]] = False
self.mqtt_client.add_callback(self.topic + '/output/%d/set' % (i + 1), self.__rx_set__)
def __rx_set__(self, client, userdata, message):
data = message.payload.decode('utf-8')
key = message.topic.split('/')[-3] + '/' + message.topic.split('/')[-2]
self.logger.info("Received set data for %s: %s", key, repr(data))
self.__set__(key, data)
self.send_device_status(key)
if key.startswith("output/"):
if data == "true":
self.power_on(key)
def send_device_status(self, key):
data = self[key]
self.logger.info("Sending status for %s: %s", key, repr(data))
self.mqtt_client.send(self.topic + '/' + key, data)

3
devices/shelly.py
View File

@ -46,6 +46,7 @@ class shelly_sw1(base):
PROPERTIES = [
"relay/0",
]
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
self["state"] = "off"
@ -62,7 +63,7 @@ class shelly_sw1(base):
self.send_device_status(key)
if key == "relay/0":
if data.lower() == "on":
self.power_on()
self.power_on(key)
if self.__auto_off__ is not None:
self.__auto_off__.run()
else:

6
devices/tradfri.py
View File

@ -53,6 +53,7 @@ class sw(base):
PROPERTIES = [
"state",
]
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
self["state"] = "off"
@ -71,12 +72,13 @@ class sw(base):
self.__set__(key, data[key])
self.send_device_status()
if "state" in data and data.get("state", 'OFF').lower() == "on":
self.power_on()
self.power_on("state")
def __rx_get__(self, client, userdata, message):
self.send_device_status()
def power_on_action(self):
self["state"] = "on"
self.send_device_status()
def send_device_status(self):
@ -95,6 +97,7 @@ class sw_br(sw):
PROPERTIES = sw.PROPERTIES + [
"brightness",
]
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
self["brightness"] = 64
@ -110,6 +113,7 @@ class sw_br_ct(sw_br):
PROPERTIES = sw_br.PROPERTIES + [
"color_temp",
]
def __init__(self, mqtt_client, topic):
super().__init__(mqtt_client, topic)
self["color_temp"] = 413

65
home_emulation.py
View File

@ -1,15 +1,15 @@
import config
import devdi
import devdi.props as props
#import function
#import json
import logging
import mqtt
import os
import report
#import subprocess
import time
# TODO: Implementation of missing devices in devices/__init__.py
# TODO: Implementation of interface for external device stimulation
logger = logging.getLogger(config.APP_NAME)
@ -30,9 +30,9 @@ if __name__ == "__main__":
password=config.MQTT_PASSWORD, name=config.APP_NAME)
#
# Smarthome Devices
# Smarthome physical Devices
#
ddi = devdi.devices(mc)
pd = devdi.physical_devices(mc)
#
# Smart Home Functionality
@ -43,55 +43,58 @@ if __name__ == "__main__":
loc = props.LOC_GFW
# DIRK
roo = props.ROO_DIR
sml = ddi.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = ddi.get(props.STG_ZGW, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml)
sml = pd.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = pd.get(props.STG_ZGW, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml, sml.PROPERTIES[0])
sml = pd.get(props.STG_MYA, loc, roo, props.FUN_MPP)
tml = pd.get(props.STG_ZGW, loc, roo, props.FUN_DEL)
sml.register_power_on_instance(tml, sml.PROPERTIES[1])
# FLOOR
roo = props.ROO_FLO
sml = ddi.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = ddi.get(props.STG_ZGW, loc, roo, props.FUN_MAL, 1)
sml.register_power_on_instance(tml)
tml = ddi.get(props.STG_ZGW, loc, roo, props.FUN_MAL, 2)
sml.register_power_on_instance(tml)
sml = pd.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = pd.get(props.STG_ZGW, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml, sml.PROPERTIES[0])
#######
# FFW #
#######
loc = props.LOC_FFW
# JULIAN
roo = props.ROO_JUL
sml = ddi.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = ddi.get(props.STG_ZFW, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml)
sml = pd.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = pd.get(props.STG_ZFW, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml, sml.PROPERTIES[0])
# LIVINGROOM
roo = props.ROO_LIV
sml = ddi.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = ddi.get(props.STG_ZFW, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml)
sml = pd.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = pd.get(props.STG_ZFW, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml, sml.PROPERTIES[0])
# SLEEP
roo = props.ROO_SLP
sml = ddi.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = ddi.get(props.STG_ZFW, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml)
sml = pd.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = pd.get(props.STG_ZFW, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml, sml.PROPERTIES[0])
#######
# FFE #
#######
loc = props.LOC_FFE
# KITCHEN
roo = props.ROO_KIT
sml = ddi.get(props.STG_SHE, loc, roo, props.FUN_CIR)
sml = pd.get(props.STG_SHE, loc, roo, props.FUN_CIR)
sml.auto_off(600)
# LIVINGROOM
roo = props.ROO_LIV
sml = ddi.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = ddi.get(props.STG_ZFE, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml)
sml = pd.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = pd.get(props.STG_ZFE, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml, sml.PROPERTIES[0])
# SLEEP
roo = props.ROO_SLP
sml = ddi.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = ddi.get(props.STG_ZFE, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml)
sml = pd.get(props.STG_SHE, loc, roo, props.FUN_MAL)
tml = pd.get(props.STG_ZFE, loc, roo, props.FUN_MAL)
sml.register_power_on_instance(tml, sml.PROPERTIES[0])
while (True):
time.sleep(1)

5
home_emulation.sh Executable file
View File

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