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

videv base class moved to base.py

Browse Source
This commit is contained in:
Dirk Alders 2023-02-22 07:34:06 +01:00
parent ce737b42d8
commit c076f516d8
4 changed files with 107 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
base.py
View File

@ -1,3 +1,4 @@
import json
import logging
try:
@ -46,3 +47,96 @@ class mqtt_base(common_base):
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.__capabilities__ = None
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)
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)
# send default data to videv interface
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)
self.__tx_capabilities__()
def __tx_capabilities__(self):
self.mqtt_client.send(self.topic + '/' + self.KEY_INFO, json.dumps(self.capabilities))
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)
@property
def capabilities(self):
if self.__capabilities__ is None:
self.__capabilities__ = {}
self.__capabilities__['__type__'] = self.__class__.__name__
for key in self.__control_dict__:
if not key in self.__capabilities__:
self.__capabilities__[key] = {}
self.__capabilities__[key]['control'] = True
for key in self.__display_dict__.values():
if not key in self.__capabilities__:
self.__capabilities__[key] = {}
self.__capabilities__[key]['display'] = True
return self.__capabilities__

2
devices/__init__.py
View File

@ -27,7 +27,7 @@ devices (DEVICES)
"""
from base import mqtt_base
from function.videv import base as videv_base
from base import videv_base
import json
import logging
import math

116
function/videv.py
View File

@ -10,9 +10,8 @@ Targets:
* No functionality should be implemented here
"""
from base import mqtt_base
from base import videv_base
from function.rooms import room, room_collection
import json
import time
try:
@ -21,100 +20,7 @@ except ImportError:
ROOT_LOGGER_NAME = 'root'
class 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.__capabilities__ = None
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)
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)
# send default data to videv interface
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)
self.__tx_capabilities__()
def __tx_capabilities__(self):
self.mqtt_client.send(self.topic + '/' + self.KEY_INFO, json.dumps(self.capabilities))
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)
@property
def capabilities(self):
if self.__capabilities__ is None:
self.__capabilities__ = {}
self.__capabilities__['__type__'] = self.__class__.__name__
for key in self.__control_dict__:
if not key in self.__capabilities__:
self.__capabilities__[key] = {}
self.__capabilities__[key]['control'] = True
for key in self.__display_dict__.values():
if not key in self.__capabilities__:
self.__capabilities__[key] = {}
self.__capabilities__[key]['display'] = True
return self.__capabilities__
class videv_switching(base):
class videv_switching(videv_base):
KEY_STATE = 'state'
def __init__(self, mqtt_client, topic, sw_device, sw_key):
@ -124,7 +30,7 @@ class videv_switching(base):
self.__tx_capabilities__()
class videv_switching_timer(base):
class videv_switching_timer(videv_base):
KEY_STATE = 'state'
KEY_TIMER = 'timer'
@ -136,7 +42,7 @@ class videv_switching_timer(base):
self.__tx_capabilities__()
class videv_switching_motion(base):
class videv_switching_motion(videv_base):
KEY_STATE = 'state'
#
KEY_TIMER = 'timer'
@ -155,7 +61,7 @@ class videv_switching_motion(base):
self.__tx_capabilities__()
class videv_switch_brightness(base):
class videv_switch_brightness(videv_base):
KEY_STATE = 'state'
KEY_BRIGHTNESS = 'brightness'
@ -167,7 +73,7 @@ class videv_switch_brightness(base):
self.__tx_capabilities__()
class videv_switch_brightness_color_temp(base):
class videv_switch_brightness_color_temp(videv_base):
KEY_STATE = 'state'
KEY_BRIGHTNESS = 'brightness'
KEY_COLOR_TEMP = 'color_temp'
@ -181,7 +87,7 @@ class videv_switch_brightness_color_temp(base):
self.__tx_capabilities__()
class videv_heating(base):
class videv_heating(videv_base):
KEY_USER_TEMPERATURE_SETPOINT = 'user_temperature_setpoint'
KEY_VALVE_TEMPERATURE_SETPOINT = 'valve_temperature_setpoint'
KEY_AWAY_MODE = 'away_mode'
@ -209,7 +115,7 @@ class videv_heating(base):
self.__tx_capabilities__()
class videv_multistate(base):
class videv_multistate(videv_base):
KEY_STATE = 'state_%d'
def __init__(self, mqtt_client, topic, key_for_device, device, num_states, default_values=None):
@ -230,7 +136,7 @@ class videv_multistate(base):
self.__tx_capabilities__()
class videv_audio_player(base):
class videv_audio_player(videv_base):
KEY_ACTIVE_PLAYER = 'player_%d'
KEY_TITLE = 'title'
NO_TITLE = '---'
@ -255,7 +161,7 @@ class videv_audio_player(base):
return self.__capabilities__
class videv_warnings(base):
class videv_warnings(videv_base):
MAX_WARNINGS = 10
KEY_WARNING = 'html_short'
@ -270,7 +176,7 @@ class videv_warnings(base):
self.__tx__(self.KEY_WARNING, txt)
class all_off(base):
class all_off(videv_base):
ALLOWED_CLASSES = (room, room_collection, )
def __init__(self, mqtt_client, topic, room_collection):

2
smart_brain.py
View File

@ -13,7 +13,7 @@ logger = logging.getLogger(config.APP_NAME)
VERS_MAJOR = 1
VERS_MINOR = 2
VERS_PATCH = 3
VERS_PATCH = 4
INFO_TOPIC = "__info__"
INFO_DATA = {