Interaction avec les devices domoticz dans un plugin Python

[vinchz31]

Bonjour à tous,

Je débute dans les plugin Python pour domoticz.
J'ai repris un plugin existant SolarEdge que j'ai adapté pour récupérer la production en KW de mes panneaux solaires. Jusqu'ici je m'en suis sorti sans difficulté en comprenant / adaptant le code.

Le code permet de manipuler sans souci les devices (compteurs KWh, etc.) qui sont automatiquement créés lors de l'ajout du matériel qui appelle ce plugin. Toutefois, maintenant que j'ai récupéré ma production électrique, je cherche à actionner automatiquement des interrupteurs (déjà présents dans Domoticz) pour déclencher mon cumulus, etc. quand j'ai un surplus de production.

Ma question est la suivante : comment peut-on récupérer les valeurs des devices domoticz créés "en dehors" du pluggin et modifier celles-ci ? Exemple, comment puis-je connaître l'état de mon interrupteur du cumulus à partir de son nom (que je passerais en paramètre du pluggin) et agir dessus ?

Auriez-vous quelques exemples pour réaliser ce type d'interaction ?
Merci d'avance pour votre aide précieuse !!!!

Très bonne journée à tous

[Keros]

C'est la même chose que pour allumer une lampe à partir d'un bouton. Il faut faire un peu de programmation. Dans ton cas, je te propose directement le LUA ou le Dzvents. Vu ce que tu fais avec le Python, je ne pense pas que le Blocky te soit utile.

Dans le principe, tu fais tourner les scripts soit en fonction du temps (toutes les minutes) soit lorsqu'un objet change de valeur et tu peux lancer des actions.
Si tu ne connais pas du tout le LUA ou le Dzvents, je te propose de regarder les petits guides que j'ai écrits dans ma présentation : il y en a un concernant les débuts en programmation.

[vinchz31]

Merci pour ta réponse Keros !
Je connais DZEvent et Lua que j'utilise par ailleurs mais j'aurais aimé tout pouvoir gérer dans mon plugin PYTHON plutôt que de le faire toutes les minutes par exemple...
N'y-a-t-il pas une possibilité de faire ces intéractions via une bibliothèque PYTHON directement ?
Très bonne soirée !

[Keros]

Pour moi, les plugins sont là pour transformer une donnée en une valeur utilisable par DomoticZ. L'idée ne me semble pas d'y mettre de l'intelligence dedans.

J'ai vu que dans le dossier Scripts de DomoticZ, il y avait un dossier Python à côté de DzVents et LUA. Peut-être que tu peux faire quelque chose là-dedans et utiliser les Python Events.

[vinchz31]

Oui, tu as peut-être raison, peut-être que je fais fausse route et que je dois déporter l'automatisation du plugin.
Du coup, je vais regarder pour externaliser ces actions.
Merci pour tes explications...
Très bonne soirée !

[Flying Domotic]

Une façon de faire serait d'utiliser les évènements LUA ou dzVents, pour détecter les changements du/des device(s) géré(s) par le plugin, et modifier d'autres devices non gérés. Utiliser les exemples dans la partie "Evènements" de "Configuration" -> "Plus d'otions".

[lost]

Oui, tu as peut-être raison, peut-être que je fais fausse route et que je dois déporter l'automatisation du plugin.

Bonjour,

Tout ce que j'ai écrit en python l'a été avant que les plugins n'existent... et donc l'import domoticz associé. Si des choses manquent de ce côté API plugin, il reste possible d'utiliser l'API HTTP/JSON, y compris sur le localhost: Ce que j'ai alors fait faute de choix! De là, on peut faire ce qui manque comme interactions.

Le problème dans ton cas, il me semble, c'est que la partie plugin impose python et que certaines choses sont plus simples en Lua qui est globalement très bien intégré à Domoticz... mais ne permet pas de coder des plugins!

Il y a la possibilité de dissocier, je l'ai fait dans mon système, mais à l'usage et niveau maintenance sur les années ne pas avoir tout localisé de manière cohérente et introduire des dépendances évitables se réfléchit: Le jour ou une évolution du langage ou de Domoticz survient, on l'aura oublié...

Le source de mon petit module regroupant les fonctions utilitaires communes:

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

Niveau utilisation, il faut inclure ce module (ou juste les fonctions/definitions utiles, ce qui concerne juste les switch ci dessous par ex.), initialiser le logger (tous les log, par défaut dans le syslog, info ou debug se font avec lui). Extrait de code exemple pour envoyer une commande à un switch:

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)
    
    (...)

Le source du module est commenté, sur les appels/variantes (en particulier, lire état avant action pour éviter les duplicats en DB, gestion pass switch éventuel)... Il y a tout ce qu'il fat pour manipuler des switch/variables utilisateur/état security panel... voir intéragir de manire générique avec les API HTTP tierces (type caméras IP).

Il faut niveau config Domoticz avoir configuré les accès http localhost sans password.

[vinchz31]

Merci à tous les deux pour vos retours.
Je m'étais développé des fonctions en Php qui me permettent de travailler sur l'ensemble des objets Domoticz (via l'interrogation JSON) et que j'appelais via LUA sur les changements d'état ou de manière régulière.
Ca rejoint vos méthodes. Je vais rester sur ce modèle que je connais bien
Très bonne soirée !