Forum Domoticz en français

Code : Tout sélectionner

"""
UPS Monitor python plugin for Domoticz

Author: Logread, adapted from the NUT utilities (see Network UPS Tools project at http://networkupstools.org/)

Version:    0.0.1: alpha
            0.0.2: beta, changed status device to Alert v.s. multilevel switch with alarm icon
            0.1.0: 1st stable version
            0.1.1: small bux fix, edit incorrect some devices labels (AC v.s. DC)
            0.1.2: code change to avoid bug that appeared with domoticz version 3.8035
                    (devices no longer can be created and updated in the same pass
            0.1.3: code cleanup
            0.2.0: added a more comprehensive list of NUT status variables. Thanks to domoticz forum user @ycahome
            0.2.1: in case of communication error or invalid data received, stop updating devices (thanks
                    to domoticz forum user @hamster) and mark the status device as "timed out" to display in red in GUI
            0.2.2: added authentication option (thanks to domoticz forum user @copernicnic)
            0.2.3: fixed incorrect indent
"""
"""
<plugin key="NUT_UPS" name="UPS Monitor" author="logread" version="0.2.3" wikilink="http://www.domoticz.com/wiki/plugins/NUT_UPS.html" externallink="http://networkupstools.org/">
    <params>
        <param field="Address" label="UPS NUT Server IP Address" width="200px" required="true" default="127.0.0.1"/>
        <param field="Port" label="Port" width="40px" required="true" default="3493"/>
        <param field="Username" label="Username" width="200px" required="false" default=""/>
        <param field="Password" label="Password" width="200px" required="false" default=""/>
        <param field="Mode1" label="UPS NUT name" width="200px" required="true" default="ups"/>
        <param field="Mode6" label="Debug" width="75px">
            <options>
                <option label="True" value="Debug"/>
                <option label="False" value="Normal"  default="true" />
            </options>
        </param>
    </params>
</plugin>
"""
import Domoticz
import telnetlib
from datetime import datetime, timedelta


class BasePlugin:

    def __init__(self):
        self.debug = False
        self.error = True
        self.timeoutversion = False
        self.timeout = False
        self.nextpoll = datetime.now()
        self.pollinterval = 60  #Time in seconds between two polls
        self.variables = {
            # key:              [device label,         unit, value, device number, used by default]
            "ups.status":       ["UPS Status",          "",  0,     1, 1],
            "battery.charge":   ["UPS Charge",          "%", None,  2, 1],
            "battery.runtime":  ["UPS Backup Time",     "s", None,  3, 1],
            "input.voltage":    ["UPS AC Input",        "V", None,  4, 0],
            "ups.load":         ["UPS Load",            "%", None,  5, 0],
            "ups.realpower":    ["UPS Power",           "W", None,  6, 0],
            "input.frequency":  ["UPS AC Frequency",   "Hz", None,  7, 0]
        }
        self.status = {
            #code       (display,           alarm level)
            "OL":       ("ONLINE",          1), # On line (mains is present)
            "OB":       ("ONBATTERY",       4), # On battery (mains is not present)
            "LB":       ("LOWBATTERY",      4), # Low battery
            "HB":       ("HIGHBATTERY",     1), # High battery
            "RB":       ("REPLACEBATTERY",  3), # The battery needs to be replaced
            "CHRG":     ("CHARGING",        1), # The battery is charging
            "DISCHRG":  ("DISCHARGING",     3), # The battery is discharging (inverter is providing load power)
            "BYPASS":   ("BYPASS",          3), # UPS bypass circuit is active - no battery protection is available
            "CAL":      ("CALLIBRATION",    1), # UPS is currently performing runtime calibration (on battery)
            "OFF":      ("OFF",             0), # UPS is offline and is not supplying power to the load
            "OVER":     ("OVERLOAD",        4), # UPS is overloaded
            "TRIM":     ("SMARTTRIM",       3), # UPS is trimming incoming voltage (called "buck" in some hardware)
            "BOOST":    ("BOOST",           3), # UPS is boosting incoming voltage
            "FSD":      ("FORCE_SHUTDOWN",  4)  # Forced Shutdown
        }
        self.statusflags = []
        self.alert = 0
        self.telnettimeout = 2  # seconds
        return


    def onStart(self):
        Domoticz.Debug("onStart called")
        if Parameters["Mode6"] == 'Debug':
            self.debug = True
            Domoticz.Debugging(1)
            DumpConfigToLog()
        else:
            Domoticz.Debugging(0)
        # create the mandatory child device if it does not yet exist
        if 1 not in Devices:
            Domoticz.Device(Name="UPS Status Mode", Unit=1, TypeName="Alert", Used=1).Create()
            Devices[1].Update(nValue=0, sValue="") # Grey icon to reflect not yet polled
        # check if our version of domoticz supports the "TimedOut" device attribute (I cannot remember which version of
        # domoticz has this implemented, so best is to check if supported by "trial and error" method)
        try:
            temp = Devices[1].TimedOut
        except AttributeError:
            self.timeoutversion = False
        else:
            self.timeoutversion = True

    def onStop(self):
        Domoticz.Debug("onStop called")
        Domoticz.Debugging(0)


    def onHeartbeat(self):
        now = datetime.now()
        if now >= self.nextpoll:
            self.nextpoll = now + timedelta(seconds=self.pollinterval)
            # poll the NUT UPS server
            try:
                nut = telnetlib.Telnet(host=Parameters["Address"], port=Parameters["Port"], timeout=5)
            except Exception as errorcode:
                Domoticz.Error("Cannot communicate with NUT Server at {}:{} due to {}".format(
                    Parameters["Address"], Parameters["Port"], errorcode.args))
                self.error = True
                self.UpdateDevice("ups.status")  # we flag the error to the status device
            else:
                # proceed with authentication if parameters are present
                if Parameters["Username"] != "" and Parameters["Password"] != "":
                    try:
                        nut.write(bytes("USERNAME %s\n" % Parameters["Username"], "utf-8"))
                        result = nut.read_until(b"\n", self.telnettimeout).decode()
                        if result[:2] != "OK":
                            raise Exception(result.replace("\n", ""))
                        nut.write(bytes("PASSWORD %s\n" % Parameters["Password"], "utf-8"))
                        result = nut.read_until(b"\n", self.telnettimeout).decode()
                        if result[:2] != "OK":
                            raise Exception(result.replace("\n", ""))
                    except Exception as errorcode:
                        Domoticz.Error("Authentication error: {}".format(errorcode.args))
                        self.error = True
                        self.UpdateDevice("ups.status")  # we flag the error to the status device
                        nut.close()
                        return
                # read NUT variables
                nut.write(bytes("LIST VAR {}\n".format(Parameters["Mode1"]), "utf-8"))
                response = nut.read_until(b"\n", self.telnettimeout).decode()
                if response == "BEGIN LIST VAR {}\n".format(Parameters["Mode1"]):
                    response = nut.read_until(bytes("END LIST VAR {}\n".format(Parameters["Mode1"]), "utf-8"),
                                              self.telnettimeout).decode()
                    offset = len("VAR {} ".format(Parameters["Mode1"]))
                    end_offset = 0 - (len("END LIST VAR %s\n" % Parameters["Mode1"]) + 1)
                    for current in response[:end_offset].split("\n"):
                        key = str(current[offset:].split('"')[0].replace(" ", ""))
                        data = current[offset:].split('"')[1]
                        if key in self.variables:
                            self.variables[key][2] = data
                    self.error = False
                else:
                    Domoticz.Error("Error reading UPS variables: {}".format(response.replace("\n", "")))
                    self.error = True
                nut.close()
                self.statusflags = []  # reset status flags list
                self.alert = 0  # reset alarm level to 0
                for key in self.variables:
                    Domoticz.Debug("Variable {} = {}".format(self.variables[key][0], self.variables[key][2]))
                    if self.variables[key][2]:  # skip any variables not reported by the NUT server
                        self.UpdateDevice(key)  # create/update the relevant child devices


    def UpdateDevice(self, key):

        # inner function to perform the actual update
        def DoUpdate(Unit=0, nValue=0, sValue="", TimedOut=0):
            try:
                if self.timeoutversion:  # Device attribute TimeOut exists in the host domoticz version or not
                    Devices[Unit].Update(nValue=nValue, sValue=sValue, TimedOut=TimedOut)
                else:
                    Devices[Unit].Update(nValue=nValue, sValue=sValue)
            except Exception as errorcode:
                Domoticz.Error("Failed to update device unit {} due to {}".format(Unit, errorcode.args))

        # Make sure that the Domoticz device still exists (they can be deleted) before updating it
        if self.variables[key][3] in Devices:
            if key == "ups.status":
                if self.error:
                    nvalue = 0
                    svalue = "COMMUNICATION ERROR"
                    timedout = True
                else:
                    # iterate over the codes in the status string
                    for word in self.variables[key][2].split(" "):
                        try:
                            temp = self.status[word]
                        except KeyError:
                            # code not in our list of possible statuses... do not "translate" but report as is
                            self.statusflags.append(str(word))
                        else:
                            # code is known so translated and alert level updated accordingly
                            self.statusflags.append(self.status[word][0])
                            self.alert = max(self.alert, self.status[word][1])
                    nvalue = self.alert
                    svalue = " ".join(self.statusflags)
                    timedout = False
                if timedout != self.timeout:
                    self.timeout = timedout
                    timedoutchange = True
                else:
                    timedoutchange = False
                if Devices[self.variables[key][3]].sValue != svalue or timedoutchange:
                    DoUpdate(self.variables[key][3], nvalue, svalue, timedout)
            else:
                if not self.error:
                    nvalue = 0
                    svalue = str(self.variables[key][2])
                    DoUpdate(self.variables[key][3], nvalue, svalue)
        elif key != "ups.status":
            Domoticz.Device(Name=self.variables[key][0], Unit=self.variables[key][3], TypeName="Custom",
                            Image= 17, Options={"Custom": "1;{}".format(self.variables[key][1])},
                            Used=self.variables[key][4]).Create()
            # Update upon next poll (recursive call to update device broken in some domoticz versions)


global _plugin
_plugin = BasePlugin()

def onStart():
    global _plugin
    _plugin.onStart()

def onStop():
    global _plugin
    _plugin.onStop()

def onHeartbeat():
    global _plugin
    _plugin.onHeartbeat()

# Plugin specific functions ---------------------------------------------------

# Generic helper functions
def DumpConfigToLog():
    for x in Parameters:
        if Parameters[x] != "":
            Domoticz.Debug( "'" + x + "':'" + str(Parameters[x]) + "'")
    Domoticz.Debug("Device count: " + str(len(Devices)))
    for x in Devices:
        Domoticz.Debug("Device:           " + str(x) + " - " + str(Devices[x]))
        Domoticz.Debug("Device ID:       '" + str(Devices[x].ID) + "'")
        Domoticz.Debug("Device Name:     '" + Devices[x].Name + "'")
        Domoticz.Debug("Device nValue:    " + str(Devices[x].nValue))
        Domoticz.Debug("Device sValue:   '" + Devices[x].sValue + "'")
        Domoticz.Debug("Device LastLevel: " + str(Devices[x].LastLevel))
    return