commit 90c29b5578cb8e4fd44c8db5029e84440ec327a4 Author: François JUMELLE Date: Tue Mar 22 15:28:38 2022 +0100 First release diff --git a/plugin.py b/plugin.py new file mode 100644 index 0000000..1a4574c --- /dev/null +++ b/plugin.py @@ -0,0 +1,205 @@ +# Author: fjumelle +# +""" + + +

Enhances Selector


+ Selector to pilot other selectors. +

Devices

+
    +
  • Multi level selector "Selector"
    • +
+

Configuration

+
    +
  • Each "Level x" field is formatted like that: + Name=[device_id, device_value, "devide_type"], [device_id, device_value, "devide_type"], ...
    • +
  • Supported "device-type" are: "Switch", "Selector" and "Thermostat".
    • +
+ + + + + + + + + + + + + + +""" +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