Forum Domoticz en français

Code : Tout sélectionner

# Domoticz utilities module
# Changelog :
#   30/05/2020, YL, 1st version, factorize code.
#   11/09/2020, Pylint check.
#   15/10/2020, Added server response checks/API coherence.
#   18/10/2020, Added generic http get simple API for IP devices (cams...).
#   01/03/2021, Adde update-only mode for switch set.
#   2023/01/01, Python3 support.

import sys
import logging
import json
import requests

dmtJsonSwitch    = {"type":"command", "param":"switchlight", "idx":999, "switchcmd":"Off", "passcode":""}
dmtJsonGetSwitch = {"type":"devices", "rid":999}

dmtJsonGetVar = {"type":"command", "param":"getuservariable", "idx":999}
dmtJsonUpdVar = {"type":"command", "param":"updateuservariable",
                 "vname":"DOMOTICZ_USR_VAR", "vtype":0, "vvalue":0}

dmtJsonGetSec = {"type":"command", "param":"getsecstatus"}

# Internal fct to validate domoticz http server status + response,
# returns json server output or None if server response != 200.
def _dmtValidateHttpResponse(dmt_rget, logger):
    try:
        if dmt_rget.status_code != 200:
            logger.log(logging.INFO, "Server error %d", dmt_rget.status_code)
            return None

        # Decode json response that should be now be valid...
        json_rget = json.loads(dmt_rget.text)

        if json_rget["status"] != "OK":
            logger.log(logging.INFO, "Server returned %s", json_rget["status"])
            return None
    except: # pylint: disable=bare-except
        logger.log(logging.INFO, "Could not validate JSON response...")
        return None

    return json_rget

# Get domoticz security panel status
def dmtGetSecStatus(url, logger):
    """
    Get Domoticz sec status, returns -1 if server response error, -2 if connection fails...
    """
    ret = -1

    try:
        # Connect to Domoticz via JSON API and send data
        dmtRget = requests.get(url, params=dmtJsonGetSec)
    except requests.exceptions.RequestException as dmtErr:
        logger.log(logging.ERROR, "Unable to connect with URL=%s \nGet requests error : %s", url, dmtErr)
        return -2

    #logger.log(logging.DEBUG, "Sent data: [%s]" % (dmtRget.url))
    #logger.log(logging.DEBUG, "dmtGetSecStatus() server response :\n%s" % (dmtRget.text))

    # Validate server response...
    jsonRget = _dmtValidateHttpResponse(dmtRget, logger)
    if jsonRget != None:
        try:
            ret = jsonRget["secstatus"]
        except: # pylint: disable=bare-except
            logger.log(logging.ERROR, "JSON response error : secstatus")
            return -1

    logger.log(logging.DEBUG, "dmtGetSecStatus() = %d", ret)

    return ret

# Get domoticz integer user variable, returns sys.maxint if error
def dmtGetUsrVarInt(url, idx, name, logger):
    """
    Get Domoticz integer user variable, returns sys.maxint whatever error is (not to confuse with valid integer return)...
    """
    ret = sys.maxsize

    dmtJsonGetVar['idx'] = idx

    try:
        # Connect to Domoticz via JSON API and send data
        dmtRget = requests.get(url, params=dmtJsonGetVar)
    except requests.exceptions.RequestException as dmtErr:
        logger.log(logging.ERROR, "Unable to connect with URL=%s \nGet requests error : %s", url, dmtErr)
        return ret

    #logger.log(logging.DEBUG, "Sent data: [%s]" % (dmtRget.url))
    #logger.log(logging.DEBUG, "dmtGetUsrVar(idx=%d <=> %s) server response :\n%s" % (idx, name, dmtRget.text))

    # Validate server response...
    jsonRget = _dmtValidateHttpResponse(dmtRget, logger)
    if jsonRget != None:
        # Check var/idx coherence to validate domotics response
        try:
            if jsonRget["result"][0]["Name"] != name:
                logger.log(logging.INFO, "dmtGetUsrVar(idx=%d <=> %s != %s) MISMATCH", idx, name, jsonRget["result"][0]["Name"])
                return ret
            ret = int(jsonRget["result"][0]["Value"])
        except: # pylint: disable=bare-except
            logger.log(logging.ERROR, "JSON response error : UsrVar Name|Value")
            return ret
        logger.log(logging.DEBUG, "dmtGetUsrVar(%s) = %d", name, ret)

    return ret

# Used to update an integer user variable.
def dmtUpdateUsrVarInt(url, name, value, logger):
    """
    Update Domoticz user variable "name" with integer value returns -1 if server response error, -2 if connection fails...
    """

    dmtJsonUpdVar['vname'] = name
    dmtJsonUpdVar['vvalue'] = value

    try:
        # Connect to Domoticz via JSON API and send data
        dmtRget = requests.get(url, params=dmtJsonUpdVar)
    except requests.exceptions.RequestException as dmtErr:
        logger.log(logging.ERROR, "Unable to connect with URL=%s \nGet requests error : %s", url, dmtErr)
        return -2

    # Validate server response...
    jsonRget = _dmtValidateHttpResponse(dmtRget, logger)
    if jsonRget is None:
        logger.log(logging.INFO, "Server error updating user var...")
        return -1

    logger.log(logging.DEBUG, "Sent data: [%s]", dmtRget.url)
    return 0

# Get domoticz switch status ; Set name to enforce idx/name matching
def dmtGetSwitch(url, idx, logger, name = None):
    """
    Get Domoticz switch status, returns -1 if idx/name mismatch or server error, -2 if connection fails...
    """
    ret = -1

    dmtJsonGetSwitch['rid'] = idx

    try:
        # Connect to Domoticz via JSON API and send data
        dmtRget = requests.get(url, params=dmtJsonGetSwitch)
    except requests.exceptions.RequestException as dmtErr:
        logger.log(logging.ERROR, "Unable to connect with URL=%s \nGet requests error : %s", url, dmtErr)
        return -2

    #logger.log(logging.DEBUG, "Sent data: [%s]" % (dmtRget.url))
    #logger.log(logging.DEBUG, "dmtGetSwitch(idx=%d <=> %s) server response :\n%s" % (idx, name, dmtRget.text))

    # Validate server response...
    jsonRget = _dmtValidateHttpResponse(dmtRget, logger)
    if jsonRget != None:
        try:
            # Check var/idx coherence to validate domoticz response
            if (name != None) and (jsonRget["result"][0]["Name"] != name):
                logger.log(logging.INFO, "dmtGetSwitch(idx=%d <=> %s != %s) MISMATCH", idx, name, jsonRget["result"][0]["Name"])
                return ret
            ret = jsonRget["result"][0]["Data"]
        except: # pylint: disable=bare-except
            logger.log(logging.ERROR, "JSON response error : Switch Name|Data")
            return ret
        logger.log(logging.DEBUG, "dmtJsonGetSwitch(%s) = %s", name, ret)

    return ret

# Set switch to On/Off ; passcode ignored if switch is not protected ; Can be set globally
# in dmtJsonSwitch dictionnary so don't modify unless filled...
def dmtSetSwitch(url, idx, switchcmd, logger, updateonly = False, passcode = None):
    """
    Set Domoticz switch idx with cmd=On/Off, returns -1 if server response error, -2 if connection fails...
    """
    ret = -1

    # In update-only mode, don't send switch command if no change...
    if updateonly:
        status = dmtGetSwitch(url, idx, logger)
        if status == switchcmd:
            logger.log(logging.DEBUG, "No change (%s) for switch idx=%d", status, idx)
            return 0

    # Update command parameters...
    dmtJsonSwitch['idx']       = idx
    dmtJsonSwitch['switchcmd'] = switchcmd
    if passcode != None:
        dmtJsonSwitch['passcode'] = passcode

    try:
        # Connect to Domoticz via JSON API and send data
        dmtRget = requests.get(url, params=dmtJsonSwitch)
    except requests.exceptions.RequestException as dmtErr:
        logger.log(logging.ERROR, "Unable to connect with URL=%s \nGet requests error %s", url, dmtErr)
        return -2

    # Validate server response...
    jsonRget = _dmtValidateHttpResponse(dmtRget, logger)
    if jsonRget != None:
        logger.log(logging.DEBUG, "Sent data: [%s]", dmtRget.url)
        ret = 0

    return ret

# Send IP device (camera...) command through it's HTTP API, returns sys.maxint if error.
# Set non 'None' user/passwd if authentification needed.
# Consecutive accesses with same URL base (IP+PORT) will keep session
def httpGet(url, user, passwd, logger):
    """
    Send camera command, returns -1 if failed...
    """
    # Keep a session for consecutive accesses as a "static"...
    if 'last_http' not in httpGet.__dict__:
        httpGet.last_http=None
        httpGet.session=None

    try:
        # Connect to cam via HTTP API and send url including command
        _auth = None
        current_http = url.split('/')[2]
        if current_http != httpGet.last_http:
            if httpGet.session is not None:
                logger.log(logging.DEBUG, "Close session for %s", httpGet.last_http)
                httpGet.session.close()
            logger.log(logging.DEBUG, "Create session for %s", current_http)
            if user is not None:
                _auth = requests.auth.HTTPDigestAuth(user, passwd)
            httpGet.session = requests.Session()
            httpGet.session.auth = _auth
        rget = httpGet.session.get(url, auth=_auth)
    except: # pylint: disable=bare-except
        logger.log(logging.INFO, "Unable to connect with URL=%s \nGet requests error!", url)
        return None

    # Request was done, so keep current
    httpGet.last_http = current_http

    logger.log(logging.DEBUG, "Sent data: [%s]", rget.url)
    logger.log(logging.DEBUG, "Server response :\n%s", rget.text)

    return rget

Code : Tout sélectionner

(...)

import logging
from dmtUtils import dmtJsonSwitch, dmtGetSwitch, dmtSetSwitch

(...)

def main():
    """
    Main
    """
    logger = logging.getLogger()
    handler = logging.StreamHandler(sys.stdout)
    
    # Configure the logger
    handler.setFormatter(
        logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        
    try:
        logLevel = 'INFO'
        #logLevel = 'DEBUG'
    except: # pylint: disable=bare-except
        print("Could not get logging_level from %s" % configFile)
        sys.exit(3)
        
    # Exemple switch command on localhost:
    url = 'http://127.0.0.1:8080/json.htm?'
    idx = MY_SWITCH_IDX
    cmd = 'On' # Or 'Off' !

    dmtSetSwitch(url, idx, cmd, logger)
    
    (...)