commit 9188c68ad91cd4e7ec008d75e76e84a8c3406021 Author: François JUMELLE Date: Mon May 24 10:32:28 2021 +0200 first commit diff --git a/plugin.py b/plugin.py new file mode 100755 index 0000000..f420b91 --- /dev/null +++ b/plugin.py @@ -0,0 +1,162 @@ +# Signal Level python plugin for Domoticz +# +# Author: fjumelle +# +""" + + +

Signal Level


+ Create devices to trace the signal level of other devices.
+
+ + + + + + + + + + + +
+""" +import Domoticz +import requests +import json +import time +import urllib.parse as parse +import urllib.request as request +from datetime import datetime, timedelta + +DEFAULT_POOLING = 60 + + +class deviceparam: + def __init__(self, unit, nvalue, svalue): + self.unit = unit + self.nvalue = nvalue + self.svalue = svalue + + +class BasePlugin: + def __init__(self): + self.debug = False + self.pooling = 30 + self.pooling_steps = 1 + self.pooling_current_step = 1 + self.devices = [] + return + + def onStart(self): + import math + + # setup the appropriate logging level + debuglevel = int(Parameters["Mode6"]) + if debuglevel != 0: + self.debug = True + Domoticz.Debugging(debuglevel) + else: + self.debug = False + Domoticz.Debugging(0) + + # Polling interval = X sec + try: + pooling = int(Parameters["Mode5"]) + except: + pooling = DEFAULT_POOLING + self.pooling_steps = math.ceil(pooling/30) + self.pooling = pooling // self.pooling_steps + Domoticz.Heartbeat(self.pooling) + + #List of devices to scan + res = DomoticzAPI("type=devices&used=true") + self.devices = self.getDevicesSignalLevel(res) + Domoticz.Debug("Scanned devices: {}".format(self.devices)) + + # create the child devices if these do not exist yet + devicecreated = [] + for idx, device in enumerate(self.devices): + idx = idx + 1 + if idx not in Devices: + Domoticz.Device(Name="{}".format(self.getName(res, device)), Unit=idx, Image=106, TypeName="Custom", Used=1).Create() + devicecreated.append(deviceparam(device, 0, "0")) + value = self.getSignalLevel(res, device) + Devices[idx].Update(nValue=int(value), sValue=str(value)) + + def onHeartbeat(self): + if self.pooling_current_step == self.pooling_steps: + res = DomoticzAPI("type=devices&used=true") + for idx, device in enumerate(self.devices): + value = self.getSignalLevel(res, device) + idx = idx + 1 + if idx in Devices: + Devices[idx].Update(nValue=int(value), sValue=str(value)) + self.pooling_current_step = 1 + else: + self.pooling_current_step = self.pooling_current_step + 1 + + def getDevicesSignalLevel(self, res): + devices = [] + for device in res['result']: + if 'SignalLevel' in device and self.is_int(device['SignalLevel']): + devices.append(int(device['idx'])) + return devices + + def getName(self, res, idx): + for device in res['result']: + if int(idx) == int(device['idx']): + return device['Name'] + return str(idx) + + def is_int(self, s): + try: + int(s) + return True + except ValueError: + return False + + def getSignalLevel(self, res, idx): + for device in res['result']: + if int(idx) == int(device['idx']): + Domoticz.Debug("Device #{} signal level: {}".format(idx, device['SignalLevel'])) + return int(device['SignalLevel']) + return 0 + +global _plugin +_plugin = BasePlugin() + +def onStart(): + global _plugin + _plugin.onStart() + +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