206 lines
8.8 KiB
Python
206 lines
8.8 KiB
Python
# Author: fjumelle
|
|
#
|
|
"""
|
|
<plugin key="EnhancedSelector" name="Enhanced Selector" author="fjumelle" version="1.0.0" wikilink="" externallink="">
|
|
<description>
|
|
<h2>Enhances Selector</h2><br/>
|
|
Selector to pilot other selectors.
|
|
<h3>Devices</h3>
|
|
<ul style="list-style-type:square">
|
|
<li>Multi level selector "Selector"</li>
|
|
</ul>
|
|
<h3>Configuration</h3>
|
|
<ul style="list-style-type:square">
|
|
<li>Each "Level x" field is formatted like that:
|
|
Name=[device_id, device_value, "devide_type"], [device_id, device_value, "devide_type"], ...</li>
|
|
<li>Supported "device-type" are: "Switch", "Selector" and "Thermostat".</li>
|
|
</ul>
|
|
</description>
|
|
<params>
|
|
<param field="Address" label="Domoticz IP Address" width="200px" required="true" default="127.0.0.1"/>
|
|
<param field="Port" label="Port" width="40px" required="true" default="8080"/>
|
|
<param field="Username" label="Domoticz Username" width="200px" required="false" default=""/>
|
|
<param field="Password" label="Domoticz Password" width="200px" required="false" default=""/>
|
|
<param field="Mode1" label="Level 1" width="500px" required="true" default=""/>
|
|
<param field="Mode2" label="Level 2" width="500px" required="false" default=""/>
|
|
<param field="Mode3" label="Level 3" width="500px" required="false" default=""/>
|
|
<param field="Mode4" label="Level 4" width="500px" required="false" default=""/>
|
|
<param field="Mode5" label="Level 5" width="500px" required="false" default=""/>
|
|
<param field="Mode6" label="Other Levels" width="500px" required="false" default=""/>
|
|
</params>
|
|
</plugin>
|
|
"""
|
|
import Domoticz
|
|
import ast
|
|
import urllib.parse as parse
|
|
import urllib.request as request
|
|
import json
|
|
|
|
DEVICE_TYPES = {
|
|
'Switch': {
|
|
'get' : 'type=devices&rid={0}',
|
|
'get_dict': 'Status',
|
|
'set' : 'type=command¶m=switchlight&idx={0}&switchcmd={1}'
|
|
},
|
|
'Selector': {
|
|
'get' : 'type=devices&rid={0}',
|
|
'get_dict': 'Level',
|
|
'set' : 'type=command¶m=switchlight&idx={0}&switchcmd=Set Level&level={1}'
|
|
},
|
|
'Thermostat': {
|
|
'get' : 'type=devices&rid={0}',
|
|
'get_dict': 'SetPoint',
|
|
'set' : 'type=command¶m=setsetpoint&idx={0}&setpoint={1}'
|
|
},
|
|
}
|
|
|
|
class BasePlugin:
|
|
def __init__(self):
|
|
self.debug = 0
|
|
self.heartbeat = 30
|
|
self.levels = 0
|
|
self.other_name = "?"
|
|
self.devices = []
|
|
self.scanned_devices = []
|
|
self.configuration_ok = True
|
|
return
|
|
|
|
def onStart(self):
|
|
#Debug level
|
|
Domoticz.Debugging(self.debug)
|
|
|
|
#Heart beat
|
|
Domoticz.Heartbeat(self.heartbeat)
|
|
|
|
#Levels definition
|
|
levels = Parameters["Mode1"]
|
|
self.levels = 1
|
|
for i in range(2, 7):
|
|
value_param = Parameters["Mode" + str(i)].strip()
|
|
if value_param != "":
|
|
self.levels = self.levels + 1
|
|
levels = levels + ";" + value_param
|
|
|
|
devices = levels.split(";")
|
|
Domoticz.Debug("Levels: " + str(self.levels))
|
|
Domoticz.Debug(str(devices))
|
|
|
|
for level in range(0, self.levels):
|
|
device = devices[level].split('=')
|
|
if len(device) != 2:
|
|
self.configuration_ok = False
|
|
Domoticz.Error("Cannot find 'name=configuration' for the 'Device' #{}: '{}'.".format(level+1, devices[level]))
|
|
try:
|
|
list_devices = ast.literal_eval("[" + device[1] + "]")
|
|
except:
|
|
self.configuration_ok = False
|
|
Domoticz.Error("Cannot find 'name=[device_id, device_value, device_type],[...]' for the 'Device' #{}: '{}'.".format(level+1, devices[level]))
|
|
for conf in range (0, len(list_devices)):
|
|
if len(list_devices[conf]) != 3:
|
|
self.configuration_ok = False
|
|
Domoticz.Error("Cannot find '[device_id, device_value, device_type]' in the configuration {} of the 'Device' #{}: '{}'.".format(conf+1, level+1, list_devices[conf]))
|
|
if not isinstance(list_devices[conf][0], int):
|
|
self.configuration_ok = False
|
|
Domoticz.Error("'device_id' in the configuration {} of the 'Device' #{} shall be an integer: '{}'".format(conf+1, level+1, list_devices[conf][0]))
|
|
if not list_devices[conf][2] in DEVICE_TYPES.keys():
|
|
self.configuration_ok = False
|
|
Domoticz.Error("'device_type' in the configuration {} of the 'Device' #{} shall be {}: '{}'".format(conf+1, level+1, DEVICE_TYPES.keys(), list_devices[conf][2]))
|
|
self.devices.append([device[0], list_devices, level*10])
|
|
#List of devices to scan
|
|
for item in list_devices:
|
|
self.scanned_devices.append("{}:{}".format(item[0], item[2]))
|
|
#Add the 'unknown' devide
|
|
self.devices.append([self.other_name, [], self.levels*10])
|
|
|
|
#Remove duplicated in the scanned_devices
|
|
self.scanned_devices = list(dict.fromkeys(self.scanned_devices))
|
|
|
|
#Create device "Selector"
|
|
if 1 not in Devices:
|
|
Options = {"LevelActions": "||",
|
|
"LevelNames": "|".join([item[0] for item in self.devices]),
|
|
"LevelOffHidden": "false",
|
|
"SelectorStyle": "0"}
|
|
Domoticz.Device(Name="Selector", Unit=1, TypeName="Selector Switch", Switchtype=18,
|
|
Options=Options, Used=1).Create()
|
|
|
|
default_value = self.devices[self.levels][2]
|
|
Devices[1].Update(nValue=int(default_value), sValue=str(default_value), TimedOut = 0)
|
|
|
|
def onCommand(self, Unit, Command, Level, Hue):
|
|
if self.configuration_ok:
|
|
idx = int(Level//10)
|
|
if Unit == 1:
|
|
Devices[1].Update(nValue=int(Level), sValue=str(Level), TimedOut = 0)
|
|
for device in self.devices[idx][1]:
|
|
did = device[0]
|
|
dvalue = device[1]
|
|
dtype = device[2]
|
|
Domoticz.Log("Device {} (type {}): Requested value {}".format(did, dtype, dvalue))
|
|
DomoticzAPI(DEVICE_TYPES[dtype]['set'].format(did, dvalue))
|
|
|
|
def onHeartbeat(self):
|
|
if self.configuration_ok:
|
|
#Get value of devices to scan
|
|
scanned_devices = {}
|
|
for item in self.scanned_devices:
|
|
did, dtype = item.split(":")
|
|
did = int(did)
|
|
res = DomoticzAPI(DEVICE_TYPES[dtype]['get'].format(did))
|
|
value = res['result'][0][DEVICE_TYPES[dtype]['get_dict']]
|
|
scanned_devices[did] = value
|
|
|
|
Domoticz.Debug("scanned:" + str(scanned_devices))
|
|
for level in self.devices:
|
|
status = True
|
|
for device in level[1]:
|
|
if device[1] != scanned_devices[device[0]]:
|
|
status = False
|
|
break
|
|
if status:
|
|
value = int(level[2])
|
|
if Devices[1].nValue != value:
|
|
Devices[1].Update(nValue=value, sValue=str(value), TimedOut = 0)
|
|
Domoticz.Debug("New status = " + str(level[0]))
|
|
return
|
|
|
|
global _plugin
|
|
_plugin = BasePlugin()
|
|
|
|
def onStart():
|
|
global _plugin
|
|
_plugin.onStart()
|
|
|
|
def onCommand(Unit, Command, Level, Hue):
|
|
global _plugin
|
|
_plugin.onCommand(Unit, Command, Level, Hue)
|
|
|
|
def onHeartbeat():
|
|
global _plugin
|
|
_plugin.onHeartbeat()
|
|
|
|
def DomoticzAPI(APICall):
|
|
resultJson = None
|
|
url = "http://{}:{}/json.htm?{}".format(Parameters["Address"], Parameters["Port"], parse.quote(APICall, safe="&="))
|
|
Domoticz.Debug("Calling domoticz API: {}".format(url))
|
|
try:
|
|
req = request.Request(url)
|
|
if Parameters["Username"] != "":
|
|
Domoticz.Debug("Add authentification for user {}".format(Parameters["Username"]))
|
|
credentials = ('%s:%s' % (Parameters["Username"], Parameters["Password"]))
|
|
encoded_credentials = base64.b64encode(credentials.encode('ascii'))
|
|
req.add_header('Authorization', 'Basic %s' % encoded_credentials.decode("ascii"))
|
|
|
|
response = request.urlopen(req)
|
|
if response.status == 200:
|
|
resultJson = json.loads(response.read().decode('utf-8'))
|
|
if resultJson["status"] != "OK":
|
|
Domoticz.Error("Domoticz API returned an error: status = {}".format(resultJson["status"]))
|
|
resultJson = None
|
|
else:
|
|
Domoticz.Error("Domoticz API: http error = {}".format(response.status))
|
|
except Exception as err:
|
|
Domoticz.Error("Error calling '{}'".format(url))
|
|
Domoticz.Error(str(err))
|
|
return resultJson
|