Forum Domoticz en français

Code : Tout sélectionner

#!/usr/bin/env python

import requests
from datetime import date
import time

# Variables pour Domoticz
domoticz_ip = 'xxx.xxx.x.xx:8080'
idx_couleur_jour = '109'
idx_couleur_jour_j1 = '110'
vname_couleur_jour = 'Couleur TEMPO J'
vname_couleur_jour_j1 = 'Couleur TEMPO J+1'

def map_code_to_color(code):
    # Mapping des codes de jour à des couleurs spécifiques
    mapping = {
        0: "Non défini",
        1: "Bleu",
        2: "Blanc",
        3: "Rouge"
    }
    return mapping.get(code, "Inconnu")

def get_tempo_colors():
    # URLs de l'API
    url_today = 'https://www.api-couleur-tempo.fr/api/jourTempo/today'
    url_tomorrow = 'https://www.api-couleur-tempo.fr/api/jourTempo/tomorrow'

    try:
        # Requête pour obtenir la couleur du jour
        response_today = requests.get(url_today)
        if response_today.status_code == 200:
            data_today = response_today.json()
            couleur_jour = map_code_to_color(data_today.get('codeJour', 0))
        else:
            log_to_domoticz(f"Erreur lors de la requête pour la couleur du jour: {response_today.status_code}")
            return None, None

        # Requête pour obtenir la couleur du lendemain
        response_tomorrow = requests.get(url_tomorrow)
        if response_tomorrow.status_code == 200:
            data_tomorrow = response_tomorrow.json()
            couleur_jour_j1 = map_code_to_color(data_tomorrow.get('codeJour', 0))
        else:
            log_to_domoticz(f"Erreur lors de la requête pour la couleur du lendemain: {response_tomorrow.status_code}")
            return couleur_jour, None

        return couleur_jour, couleur_jour_j1

    except Exception as e:
        log_to_domoticz(f"Une erreur s'est produite: {e}")
        return None, None

def update_domoticz_variables(couleur_jour, couleur_jour_j1):
    # URL de l'API JSON de Domoticz
    url_domoticz = f'http://{domoticz_ip}/json.htm'

    # Paramètres pour mettre à jour la variable de couleur du jour (J) dans Domoticz
    params_couleur_jour = {'type': 'command', 'param': 'updateuservariable', 'idx': idx_couleur_jour, 'vname': vname_couleur_jour, 'vtype': 2, 'vvalue': couleur_jour}

    # Paramètres pour mettre à jour la variable de couleur du jour suivant (J+1) dans Domoticz
    params_couleur_jour_j1 = {'type': 'command', 'param': 'updateuservariable', 'idx': idx_couleur_jour_j1, 'vname': vname_couleur_jour_j1, 'vtype': 2, 'vvalue': couleur_jour_j1}

    try:
        # Envoyer la requête pour mettre à jour la variable de couleur du jour (J) dans Domoticz
        response_couleur_jour = requests.get(url_domoticz, params=params_couleur_jour)

        # Vérifier si la requête a réussi
        if response_couleur_jour.status_code != 200:
            log_to_domoticz(f"Erreur lors de la mise à jour de la couleur TEMPO J dans Domoticz: {response_couleur_jour.status_code}")

        # Envoyer la requête pour mettre à jour la variable de couleur du jour suivant (J+1) dans Domoticz
        response_couleur_jour_j1 = requests.get(url_domoticz, params=params_couleur_jour_j1)

        # Vérifier si la requête a réussi
        if response_couleur_jour_j1.status_code != 200:
            log_to_domoticz(f"Erreur lors de la mise à jour de la couleur TEMPO J+1 dans Domoticz: {response_couleur_jour_j1.status_code}")

    except Exception as e:
        log_to_domoticz(f"Une erreur s'est produite lors de la mise à jour des variables dans Domoticz: {e}")

def log_to_domoticz(message):
    # URL pour ajouter un log dans Domoticz
    url_log = f'http://{domoticz_ip}/json.htm?type=command&param=addlogmessage&message={message}'

    try:
        response = requests.get(url_log)
        if response.status_code != 200:
            print(f"Erreur lors de l'envoi du log à Domoticz: {response.status_code}")
    except Exception as e:
        print(f"Une erreur s'est produite lors de l'envoi du log à Domoticz: {e}")

# Appel de la fonction pour obtenir les couleurs Tempo du jour en cours et du jour suivant
couleur_jour, couleur_jour_j1 = get_tempo_colors()

# Affichage des résultats et enregistrement des logs
if couleur_jour is not None and couleur_jour_j1 is not None:
    log_to_domoticz("* Les valeurs EDF Tempo ont bien été mises à jour")
    log_to_domoticz(f"* Aujourd'hui est un jour : {couleur_jour}")
    log_to_domoticz(f"* Demain est un jour : {couleur_jour_j1}")

    # Appel de la fonction pour mettre à jour les variables dans Domoticz
    update_domoticz_variables(couleur_jour, couleur_jour_j1)
else:
    log_to_domoticz("Impossible d'obtenir les couleurs Tempo.")

# Attendre 3 secondes avant de fermer
time.sleep(3)

Code : Tout sélectionner

-- Script de récupération des informations Tempo J et J+1
-- À partir de l'API RTE en priorité (https://data.rte-france.com)
-- et les données du site www.api-couleur-tempo.fr en secours

return {
    active = true,
    on = { timer = { 'at 06:15', 'at 11:15' } },

    execute = function(dz)

        ---------------------------------------
        -- Configuration des devices (IDX)
        ---------------------------------------
        local TEMPO_AUJOURDHUI_IDX = 999  -- À REMPLACER par votre IDX
        local TEMPO_DEMAIN_IDX = 999      -- À REMPLACER par votre IDX

        dz.log("⏳ Récupération des couleurs Tempo via RTE (priorité) + fallback Tiers...", dz.LOG_INFO)

        ---------------------------------------
        -- Fonctions utilitaires
        ---------------------------------------
        
        -- Détection automatique du fuseau horaire (heure d'été/hiver)
        local function getFrenchTimezone()
            -- Méthode simple : comparer l'heure locale et UTC
            -- En hiver : UTC+1, en été : UTC+2
            local now = os.time()
            local utcDate = os.date("!*t", now)  -- Date UTC
            local localDate = os.date("*t", now) -- Date locale
            
            -- Calculer la différence en heures
            local utcTime = utcDate.hour * 3600 + utcDate.min * 60 + utcDate.sec
            local localTime = localDate.hour * 3600 + localDate.min * 60 + localDate.sec
            local diff = (localTime - utcTime) / 3600
            
            -- Gérer le passage de minuit
            if diff < -12 then diff = diff + 24 end
            if diff > 12 then diff = diff - 24 end
            
            -- En France : +1 en hiver, +2 en été
            local offset = math.floor(diff + 0.5)
            
            -- RTE utilise l'heure française : UTC+2 en été, UTC+1 en hiver
            return string.format("+%02d:00", offset)
        end
        
        local function codeToLevelRTE(val)
            if val == "BLUE" then return 10, "BLEU"
            elseif val == "WHITE" then return 20, "BLANC"
            elseif val == "RED" then return 30, "ROUGE"
            else return 0, "Pas d'info"
            end
        end

        local function codeToLevelTiers(code)
            code = tonumber(code)
            if code == 1 then return 10, "BLEU"
            elseif code == 2 then return 20, "BLANC"
            elseif code == 3 then return 30, "ROUGE"
            else return 0, "Pas d'info"
            end
        end

        -- Lecture de la clé API depuis un fichier
        local function loadAPIKey(path)
            local f, err = io.open(path, "r")
            if not f then
                return nil, "❌ Impossible de charger la clé API RTE : " .. (err or "")
            end
            local raw = f:read("*a")
            f:close()
            if not raw or raw == "" then
                return nil, "❌ Clé API RTE vide après lecture"
            end
            -- Nettoyage espaces / retours chariot éventuels
            local key = raw:gsub("^%s+", ""):gsub("%s+$", "")
            return key
        end

        local function getToken(apiKey)
            local maxRetries = 2
            local retryDelay = 3
            
            for attempt = 1, maxRetries do
                local handle = io.popen(
                    "curl -s -X POST 'https://digital.iservices.rte-france.com/token/oauth/' " ..
                    "-H 'Content-Type: application/x-www-form-urlencoded' " ..
                    "-H 'Authorization: Basic " .. apiKey .. "' " ..
                    "-d 'grant_type=client_credentials'"
                )
                local result = handle:read("*a")
                handle:close()
                
                local data = dz.utils.fromJSON(result)
                if data and data.access_token then
                    if attempt > 1 then
                        dz.log("✅ Token RTE obtenu après " .. attempt .. " tentative(s)", dz.LOG_INFO)
                    end
                    return data.access_token
                end
                
                if attempt < maxRetries then
                    dz.log("⚠️ Échec obtention token RTE (tentative " .. attempt .. "/" .. maxRetries .. "), réessai dans " .. retryDelay .. "s", dz.LOG_WARNING)
                    os.execute("sleep " .. retryDelay)
                end
            end
            
            dz.log("❌ Impossible d'obtenir un token RTE après " .. maxRetries .. " tentatives", dz.LOG_ERROR)
            return nil
        end

        local function getRTEData(token, start_date, end_date)
            local url = "https://digital.iservices.rte-france.com/open_api/tempo_like_supply_contract/v1/tempo_like_calendars" ..
                        "?start_date=" .. start_date ..
                        "&end_date=" .. end_date ..
                        "&fallback_status=true"

            local cmd = "curl -s -X GET '" .. url .. "' " ..
                        "-H 'Accept: application/json' " ..
                        "-H 'Authorization: Bearer " .. token .. "'"

            local handle = io.popen(cmd)
            local result = handle:read("*a")
            handle:close()
            return dz.utils.fromJSON(result)
        end

        local function getTiersData()
            local urls = {
                ["Tempo Aujourd'hui"] = "https://www.api-couleur-tempo.fr/api/jourTempo/today",
                ["Tempo Demain"]      = "https://www.api-couleur-tempo.fr/api/jourTempo/tomorrow"
            }
            local out = {}
            for name, url in pairs(urls) do
                local handle = io.popen("curl -s '" .. url .. "'")
                local result = handle:read("*a")
                handle:close()
                local data = dz.utils.fromJSON(result)
                if data and data.codeJour then
                    local lvl, couleur = codeToLevelTiers(data.codeJour)
                    out[name] = { level = lvl, couleur = couleur }
                else
                    dz.log("⚠️ Donnée Tiers manquante pour " .. name, dz.LOG_WARNING)
                end
            end
            return out
        end

        ---------------------------------------
        -- Récupération RTE
        ---------------------------------------
        local apiKey, err = loadAPIKey("/chemin/vers/votre/rte_api.key")  -- À MODIFIER
        if not apiKey then
            dz.log(err, dz.LOG_ERROR)
        end
        
        local token = getToken(apiKey)
        local dataRTE = {}
        
        if token then
            -- Détection automatique du timezone français
            local timezone = getFrenchTimezone()
            local today = os.date("%Y-%m-%dT00:00:00") .. timezone
            local afterTomorrow = os.date("%Y-%m-%dT00:00:00", os.time() + 2*24*3600) .. timezone

            dz.log("🔍 Plage demandée à RTE: " .. today .. " → " .. afterTomorrow, dz.LOG_INFO)

            local json = getRTEData(token, today, afterTomorrow)
            if json and json.tempo_like_calendars and json.tempo_like_calendars.values then
                dz.log("🔍 Données RTE reçues : " .. #json.tempo_like_calendars.values .. " entrée(s)", dz.LOG_INFO)
                for _, entry in ipairs(json.tempo_like_calendars.values) do
                    local lvl, couleur = codeToLevelRTE(entry.value)
                    local startDate = entry.start_date:sub(1,10)
                    local todayDate = os.date("%Y-%m-%d")
                    local tomorrowDate = os.date("%Y-%m-%d", os.time() + 24*3600)

                    if startDate == todayDate then
                        dataRTE["Tempo Aujourd'hui"] = { level = lvl, couleur = couleur }
                    elseif startDate == tomorrowDate then
                        dataRTE["Tempo Demain"] = { level = lvl, couleur = couleur }
                    end
                end
                -- Log récapitulatif de ce qui a été trouvé dans RTE
                if dataRTE["Tempo Aujourd'hui"] then
                    dz.log("✅ RTE fournit Tempo Aujourd'hui: " .. dataRTE["Tempo Aujourd'hui"].couleur, dz.LOG_INFO)
                else
                    dz.log("⚠️ RTE ne fournit PAS Tempo Aujourd'hui", dz.LOG_WARNING)
                end
                if dataRTE["Tempo Demain"] then
                    dz.log("✅ RTE fournit Tempo Demain: " .. dataRTE["Tempo Demain"].couleur, dz.LOG_INFO)
                else
                    dz.log("⚠️ RTE ne fournit PAS Tempo Demain", dz.LOG_WARNING)
                end
            else
                dz.log("⚠️ Réponse RTE invalide ou vide", dz.LOG_WARNING)
            end
        end

        ---------------------------------------
        -- Récupération Tiers (fallback)
        ---------------------------------------
        local dataTiers = getTiersData()

        ---------------------------------------
        -- Choix final et mise à jour (RTE prioritaire, puis Tiers)
        ---------------------------------------
        local finalVals = {}
        local sourcesUsed = {}

        -- Table de correspondance nom logique -> IDX
        local devices = {
            ["Tempo Aujourd'hui"] = TEMPO_AUJOURDHUI_IDX,
            ["Tempo Demain"] = TEMPO_DEMAIN_IDX
        }

        for _, deviceName in ipairs({ "Tempo Aujourd'hui", "Tempo Demain" }) do
            local src = nil
            local couleur = nil
            local level = 0

            -- Priorité RTE, fallback Tiers
            if dataRTE[deviceName] then
                src = "RTE"
                level = dataRTE[deviceName].level
                couleur = dataRTE[deviceName].couleur
            elseif dataTiers[deviceName] then
                src = "Tiers (fallback)"
                level = dataTiers[deviceName].level
                couleur = dataTiers[deviceName].couleur
            end

            -- Appliquer / logger
            if src then
                dz.log("🎯 " .. deviceName .. " : " .. (couleur or "inconnu") .. " (source = " .. src .. ")", dz.LOG_INFO)
                
                -- Mise à jour du device par son IDX (plus fiable que par nom)
                local deviceIdx = devices[deviceName]
                dz.devices(deviceIdx).switchSelector(level)
            else
                dz.log("❌ Aucune donnée disponible pour " .. deviceName, dz.LOG_ERROR)
            end

            finalVals[deviceName] = { src = src, couleur = couleur, level = level }
            if src then
                table.insert(sourcesUsed, src)
            end

            -- Comparaison si les deux sources sont présentes pour ce device
            if dataRTE[deviceName] and dataTiers[deviceName] then
                if dataRTE[deviceName].couleur ~= dataTiers[deviceName].couleur then
                    dz.log("⚠️ DIVERGENCE : RTE = " .. dataRTE[deviceName].couleur ..
                           " / Tiers = " .. dataTiers[deviceName].couleur, dz.LOG_ERROR)
                else
                    dz.log("✅ Cohérence confirmée pour " .. deviceName, dz.LOG_INFO)
                end
            end
        end  -- fin for

        -- Log récapitulatif des sources utilisées
        if #sourcesUsed > 0 then
            local uniqueSources = {}
            for _, s in ipairs(sourcesUsed) do
                uniqueSources[s] = true
            end
            local sourcesList = {}
            for s in pairs(uniqueSources) do
                table.insert(sourcesList, s)
            end
            dz.log("📊 Sources utilisées : " .. table.concat(sourcesList, ", "), dz.LOG_INFO)
        end

        dz.log("------ Fin récupération Tempo", dz.LOG_INFO)
    end
}