domoticz-SignalLevel/plugin.py

# Signal Level python plugin for Domoticz
#
# Author: fjumelle
#
"""
<plugin key="SignalLevel" name="Signal Level" author="fjumelle" version="1.0.0" wikilink="" externallink="">
    <description>
        <h2>Signal Level</h2><br/>
        Create devices to trace the signal level of other devices.<br/> 
    </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="Mode5" label="Polling interval (s)" width="40px" required="true" default="60"/>
        <param field="Mode6" label="Logging Level" width="200px">
            <options>
                <option label="Normal" value="0" default="true"/>
                <option label="Verbose" value="1"/>
            </options>
        </param>
    </params>
</plugin>
"""
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