Partage
  • Partager sur Facebook
  • Partager sur Twitter

Requêtes python à API dans boucle infini

Sujet résolu
    28 novembre 2021 à 19:14:16

    Bonjour,

    j'ai mis en place tout un système pour me permettre de contrôler mes volets même si je ne suis pas chez moi. Tout cela repose sur : Domoticz (qui gère l'exécution des relais qui commande les volets), un site internet qui possède toutes les 'boutons' (fermer et ouvrir les volets, créer des programmes...) et pour faire le lien entre les deux, j'ai créé un script python qui récupérer les données d'une api en ligne qui a les commandes (ex : shutter1Position = open) et qui exécute les commande sur Domoticz (ex : <adress_ip>:8080/json.html?idx=1&switchcmd=On).

    Pour que les commandes se passe en temps réel (quand j'appuie sur une commande sur le site web, le volet se ferme chez moi) j'ai dans le script python une boucle infini (while True) qui utilise request pour faire des requêtes à l'API pour récupérer les commandes et les exécuter.

    Malheureusement j'ai des problèmes, quelques minutes après avoir lancer le programme sur mon RPI, j'ai l'erreur suivant qui s'affiche :

    urllib3.exceptions.NewConnectionError: <urllib3.connection.HTTPConnection object at 0x75b1de50>: Failed to establish a new connection: [Errno 101] Network is unreachable

    Voici mon code (simplifié) :

    from requests import ReadTimeout, ConnectTimeout, HTTPError, Timeout, ConnectionError
    import requests
    import time
    import datetime
    import threading
    from datetime import datetime
    
    def getIdxWithName(name: str) -> str:
        dataResponse = requests.post('<adress_ip>:8080/json.htm?type=devices&filter=light&used=true&order=Name')
        data = dataResponse.json()['result']
    
        for i in range(0, len(data)):
            if data[i]['Name'] == name:
                return data[i]['idx']
        return 'not found'
    
    def deviceControl(idx: int, status: str) -> None:
        requests.get('<adress_ip>:8080/json.htm?type=command&param=switchlight&idx=' + idx + '&switchcmd=' + status)
    
    def down(shutter: int, duration: int) -> None:
        idx = getIdxWithName(str('volet' + str(shutter) + 'down'))
        deviceControl(idx, 'Off')
    
        time.sleep(int(duration))
    
        deviceControl(idx, 'On')
    
    def up(shutter: int, duration: int) -> None:
        idx = getIdxWithName(str('volet' + str(shutter) + 'up'))
        deviceControl(idx, 'Off')
    
        time.sleep(int(duration))
    
        deviceControl(idx, 'On')
    
    def dataUpdate(name, value):
        dataQuery = {'cmd':'update', 'db': 'shutters', 'name': name, 'value': value}
        requests.post(url, data=dataQuery)
    
    cache = {'shutter1': 'open', 'shutter2': 'open', 'shutter3': 'open', 'shutter4': 'open'}
    executionVocal = False
    
    while True:
        now = datetime.now()
        url = '<api_url>'
    
        dataQuery = {'cmd':'read', 'db': 'shutters'}
        dataResponse = requests.post(url, data=dataQuery)
        if (
            dataResponse.status_code != 204 and
            dataResponse.headers["content-type"].strip().startswith("application/json")
        ):
            try:
                data = dataResponse.json()['results'][0]
            except ValueError:
                print(ValueError)
    
        # manual activation
        for i in range(1, 5):
            if cache['shutter' + str(i)] != data['shutter' + str(i) + 'Position']:
                if cache['shutter' + str(i)] == 'open':
                    if data['shutter' + str(i) + 'Position'] == 'close':
                        command = threading.Thread(target=down, args=(i, 24))
                        command.start()
                    if data['shutter' + str(i) + 'Position'] == 'md':
                        command = threading.Thread(target=down, args=(i, 10))
                        command.start()
                    if data['shutter' + str(i) + 'Position'] == 'nc':
                        command = threading.Thread(target=down, args=(i, 19))
                        command.start()
                if cache['shutter' + str(i)] == 'close':
                    if data['shutter' + str(i) + 'Position'] == 'open':
                        command = threading.Thread(target=up, args=(i, 27))
                        command.start()
                    if data['shutter' + str(i) + 'Position'] == 'md':
                        command = threading.Thread(target=up, args=(i, 16))
                        command.start()
                    if data['shutter' + str(i) + 'Position'] == 'nc':
                        command = threading.Thread(target=up, args=(i, 4))
                        command.start()
                if cache['shutter' + str(i)] == 'md':
                    if data['shutter' + str(i) + 'Position'] == 'open':
                        command = threading.Thread(target=up, args=(i, 10))
                        command.start()
                    if data['shutter' + str(i) + 'Position'] == 'nc':
                        command = threading.Thread(target=down, args=(i, 10))
                        command.start()
                    if data['shutter' + str(i) + 'Position'] == 'close':
                        command = threading.Thread(target=down, args=(i, 16))
                        command.start()
                if cache['shutter' + str(i)] == 'nc':
                    if data['shutter' + str(i) + 'Position'] == 'open':
                        command = threading.Thread(target=up, args=(i, 19))
                        command.start()
                    if data['shutter' + str(i) + 'Position'] == 'md':
                        command = threading.Thread(target=up, args=(i, 10))
                        command.start()
                    if data['shutter' + str(i) + 'Position'] == 'close':
                        command = threading.Thread(target=down, args=(i, 4))
                        command.start()
    
        # program activation
        if data['programActivationSwitch'] == 'on':
            
            if now.hour == int(data['closeStartHour']) and now.minute == int(data['closeStartMinute']):
                for i in range(1, 5):
                    dataUpdate('shutter' + str(i) + 'Position', 'close')
    
            if now.hour == int(data['closeEndHour']) and now.minute == int(data['closeEndMinute']):
                for i in range(1, 5):
                    dataUpdate('shutter' + str(i) + 'Position', 'open')
    
        cache = {'shutter1': str(data['shutter1Position']), 'shutter2': str(data['shutter2Position']), 'shutter3': str(data['shutter3Position']), 'shutter4': str(data['shutter4Position'])}
    
        time.sleep(1)

    Merci pour vos réponses et vos conseils ^^

    • Partager sur Facebook
    • Partager sur Twitter
      29 novembre 2021 à 15:54:06

      Le message d'erreur signale un problème réseau.

      C'est un peu comme quand on essaie de joindre quelqu'un au téléphone, çà sonne dans le vide, ou occupé ou.... et on ré-essaie plus tard.

      • Partager sur Facebook
      • Partager sur Twitter
        29 novembre 2021 à 16:51:14

        donc à gérer avec un try/except/else au niveau du requests.post
        • Partager sur Facebook
        • Partager sur Twitter

        Requêtes python à API dans boucle infini

        × Après avoir cliqué sur "Répondre" vous serez invité à vous connecter pour que votre message soit publié.
        • Editeur
        • Markdown