Forum Domoticz en français

Code : Tout sélectionner

[#!/usr/bin/python3
# -*- coding: utf-8 -*-

"""
VicareBridge — Bridge entre l'API ViCare (Viessmann) et Domoticz.

Usage :
  Sans argument  → synchronise les données ViCare vers Domoticz
  dhw <temp>     → règle la consigne ECS (ex: python3 script.py dhw 55)
  <level>        → change le mode (0=standby 10=normal 20=comfort 30=eco 40=dhw seul)
  <temp_float>   → change la consigne de température du programme actif

Configuration :
  Créer un fichier .env dans le même dossier avec :
    VICARE_USER=votre@email.com
    VICARE_PASS=VotreMotDePasse
    VICARE_CLIENT_ID=VotreClientId
    DOMOTICZ_URL=http://192.168.1.xx:8080   (optionnel)
    TOKEN_FILE=/home/pi/.vicare_token.json  (optionnel)
"""

import os
import sys
import time
import logging

import requests
from dotenv import load_dotenv
from PyViCare.PyViCare import PyViCare

# ---------------------------------------------------------------------------
# Configuration via .env (jamais de credentials en dur dans le code)
# ---------------------------------------------------------------------------
load_dotenv()

USERNAME     = os.environ.get("VICARE_USER", "")
PASSWORD     = os.environ.get("VICARE_PASS", "")
CLIENT_ID    = os.environ.get("VICARE_CLIENT_ID", "")
TOKEN_FILE   = os.environ.get("TOKEN_FILE", "/home/pi/domoticz/scripts/python/token.json")
DOMOTICZ_URL = os.environ.get("DOMOTICZ_URL", "http://192.168.1.45:9090")

# Index Domoticz des capteurs/actionneurs
IDX_SELECTEUR  = "599"
IDX_CHAUFFAGE  = "600"
IDX_EAU_CHAUDE = "601"
IDX_TEMP_EXT   = "593"

# ---------------------------------------------------------------------------
# Constantes métier — plus de magic strings éparpillées
# ---------------------------------------------------------------------------
class Mode:
    STANDBY  = "standby"
    DHW      = "dhw"
    DHW_HEAT = "dhwAndHeating"

class Level:
    STANDBY = 0
    NORMAL  = 10
    COMFORT = 20
    ECO     = 30
    DHW     = 40

LEVEL_TO_PROGRAM: dict[int, str] = {
    Level.NORMAL:  "normal",
    Level.COMFORT: "comfort",
    Level.ECO:     "eco",
}

PROG_TO_LEVEL: dict[str, int] = {v: k for k, v in LEVEL_TO_PROGRAM.items()}

MODE_TO_LEVEL: dict[str, int] = {
    Mode.STANDBY: Level.STANDBY,
    Mode.DHW:     Level.DHW,
}

# Chemins API ViCare
PATH_MODE_ACTIVE  = "heating.circuits.0.operating.modes.active"
PATH_ROOM_TEMP    = "heating.circuits.0.sensors.temperature.room"
PATH_OUTSIDE_TEMP = "heating.sensors.temperature.outside"
PATH_DHW_STORAGE  = "heating.dhw.sensors.temperature.hotWaterStorage"
PATH_DHW_SETPOINT = "heating.dhw.temperature.main"
PATH_PROGRAM      = "heating.circuits.0.operating.programs.{prog}"

# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Classe principale
# ---------------------------------------------------------------------------
class VicareBridge:
    """Bridge entre l'API ViCare et Domoticz."""

    def __init__(self) -> None:
        self._check_credentials()
        logger.info("Connexion à l'API ViCare...")
        try:
            self.vicare = PyViCare()
            self.vicare.initWithCredentials(USERNAME, PASSWORD, CLIENT_ID, TOKEN_FILE)

            if not self.vicare.devices:
                raise RuntimeError("Aucun device ViCare trouvé")

            # Préférer devices[1] si disponible (cas multi-équipements)
            try:
                self.device = self.vicare.devices[1]
            except IndexError:
                logger.warning("devices[1] indisponible → fallback sur devices[0]")
                self.device = self.vicare.devices[0]

            self.service = self.device.service
            self.circuit  = self.device.asGazBoiler().getCircuit(0)
            logger.info("Connexion réussie.")

        except Exception as e:
            logger.error(f"Erreur d'initialisation : {e}")
            sys.exit(1)

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _check_credentials() -> None:
        """Vérifie que les variables d'environnement obligatoires sont présentes."""
        missing = [v for v in ("VICARE_USER", "VICARE_PASS", "VICARE_CLIENT_ID") if not os.environ.get(v)]
        if missing:
            logger.error(f"Variables d'environnement manquantes : {', '.join(missing)}")
            logger.error("Créez un fichier .env avec VICARE_USER, VICARE_PASS, VICARE_CLIENT_ID")
            sys.exit(1)

    def safe_get(self, path: str, default=None):
        """Lecture sécurisée d'une propriété ViCare."""
        try:
            data = self.service.getProperty(path)
            return data["properties"]["value"]["value"]
        except Exception as e:
            logger.warning(f"Lecture échouée ({path}) : {e}")
            return default

    def update_domoticz(self, idx: str, svalue, nvalue: int = 0, retries: int = 3) -> bool:
        """
        Met à jour un device Domoticz.
        Réessaie jusqu'à `retries` fois en cas d'échec réseau.
        Retourne True si succès, False sinon.
        """
        url = f"{DOMOTICZ_URL}/json.htm"

        if idx == IDX_SELECTEUR:
            params = {
                "type":      "command",
                "param":     "switchlight",
                "idx":       idx,
                "switchcmd": "Set Level",
                "level":     svalue,
            }
        else:
            params = {
                "type":   "command",
                "param":  "udevice",
                "idx":    idx,
                "nvalue": nvalue,
                "svalue": str(svalue),
            }

        for attempt in range(1, retries + 1):
            try:
                r = requests.get(url, params=params, timeout=5)
                r.raise_for_status()
                # Vérification du statut applicatif Domoticz
                data = r.json()
                if data.get("status") != "OK":
                    logger.warning(f"Domoticz IDX {idx} — statut inattendu : {data.get('status')}")
                return True
            except Exception as e:
                if attempt == retries:
                    logger.error(f"Domoticz IDX {idx} inaccessible après {retries} tentatives : {e}")
                else:
                    logger.warning(f"Domoticz IDX {idx} — tentative {attempt}/{retries} échouée : {e}")
                    time.sleep(1)

        return False

    def wait_for_mode(self, expected_mode: str, timeout: int = 10) -> bool:
        """
        Attend que le mode actif de la chaudière corresponde à `expected_mode`.
        Remplace le sleep fixe de 1.5 s par une vérification active.
        Retourne True si le mode est confirmé, False si timeout.
        """
        for i in range(timeout):
            current = self.safe_get(PATH_MODE_ACTIVE)
            if current == expected_mode:
                logger.info(f"Mode confirmé : {expected_mode} (après {i+1}s)")
                return True
            time.sleep(1)
        logger.warning(f"Mode {expected_mode} non confirmé après {timeout}s (actuel : {current!r})")
        return False

    # ------------------------------------------------------------------
    # Synchronisation ViCare → Domoticz
    # ------------------------------------------------------------------

    def sync(self) -> None:
        """Lit l'état de la chaudière et pousse les valeurs vers Domoticz."""
        logger.info("Début de la synchronisation...")

        # Température extérieure
        t_ext = self.safe_get(PATH_OUTSIDE_TEMP, 0)
        self.update_domoticz(IDX_TEMP_EXT, t_ext)

        # Eau chaude sanitaire
        t_ballon  = self.safe_get(PATH_DHW_STORAGE, 0)
        t_con_ecs = self.safe_get(PATH_DHW_SETPOINT, 0)
        self.update_domoticz(IDX_EAU_CHAUDE, f"{t_ballon};{t_con_ecs}")

        # Chauffage
        prog        = self.circuit.getActiveProgram()
        mode_global = self.safe_get(PATH_MODE_ACTIVE, "unknown")
        t_amb       = self.safe_get(PATH_ROOM_TEMP, 20.0)

        if mode_global == Mode.DHW or prog == Mode.STANDBY:
            t_con_ch = 0.0
        else:
            try:
                t_con_ch = self.service.getProperty(
                    PATH_PROGRAM.format(prog=prog)
                )["properties"]["temperature"]["value"]
            except Exception as e:
                logger.warning(f"Consigne chauffage indisponible : {e}")
                t_con_ch = 0.0

        self.update_domoticz(IDX_CHAUFFAGE, f"{t_amb};{t_con_ch}")

        # Sélecteur de mode (dict simplifié, plus de if/elif en cascade)
        level = MODE_TO_LEVEL.get(mode_global, PROG_TO_LEVEL.get(prog, Level.NORMAL))
        self.update_domoticz(IDX_SELECTEUR, level)

        logger.info(f"✅ Sync OK — Mode: {mode_global} | Programme: {prog} | Niveau: {level}")

    # ------------------------------------------------------------------
    # Actions Domoticz → ViCare
    # ------------------------------------------------------------------

    def _set_dhw_temperature(self, args: list[str]) -> None:
        """Règle la consigne ECS. Argument : température en °C."""
        if len(args) < 2:
            logger.error("Commande 'dhw' : température manquante (ex: dhw 55)")
            return
        try:
            temp = int(float(args[1]))
            logger.info(f"Réglage ECS à {temp}°C")
            self.service.setProperty(
                PATH_DHW_SETPOINT,
                "setTargetTemperature",
                {"temperature": temp},
            )
        except Exception as e:
            logger.error(f"Erreur réglage ECS : {e}")

    def _set_mode(self, level: int) -> None:
        """
        Change le mode de fonctionnement de la chaudière selon le niveau Domoticz.
          0  → standby
          10 → dhwAndHeating + normal
          20 → dhwAndHeating + comfort
          30 → dhwAndHeating + eco
          40 → dhw seul
        """
        # Désactivation préalable des programmes optionnels (comfort/eco)
        for prog in ("comfort", "eco"):
            try:
                self.service.setProperty(
                    PATH_PROGRAM.format(prog=prog), "deactivate", {}
                )
            except Exception as e:
                logger.debug(f"Désactivation {prog} ignorée : {e}")

        if level == Level.STANDBY:
            self.service.setProperty(PATH_MODE_ACTIVE, "setMode", {"mode": Mode.STANDBY})

        elif level == Level.DHW:
            # ECS seul : on tente d'activer standby sur le circuit avant de basculer le mode
            try:
                self.circuit.activateProgram(Mode.STANDBY)
            except Exception:
                pass
            self.service.setProperty(PATH_MODE_ACTIVE, "setMode", {"mode": Mode.DHW})

        else:
            # Chauffage + ECS
            self.service.setProperty(PATH_MODE_ACTIVE, "setMode", {"mode": Mode.DHW_HEAT})
            # Attente active de confirmation du mode (remplace le sleep fixe)
            self.wait_for_mode(Mode.DHW_HEAT)
            program = LEVEL_TO_PROGRAM.get(level)
            if program:
                self.circuit.activateProgram(program)
            else:
                logger.warning(f"Niveau {level} inconnu, programme non activé")

        logger.info(f"Commande niveau {level} envoyée.")

    def _set_temperature(self, cmd: str) -> None:
        """Change la consigne du programme actif (argument : température flottante)."""
        try:
            temp        = float(cmd)
            prog        = self.circuit.getActiveProgram()
            mode_global = self.safe_get(PATH_MODE_ACTIVE)

            if mode_global != Mode.DHW_HEAT:
                logger.warning(f"Changement de consigne ignoré : mode actuel = {mode_global!r}")
                return

            self.service.setProperty(
                PATH_PROGRAM.format(prog=prog),
                "setTemperature",
                {"targetTemperature": temp},
            )
            logger.info(f"Consigne programme {prog!r} → {temp}°C")
        except ValueError:
            logger.error(f"Argument invalide (température attendue) : {cmd!r}")
        except Exception as e:
            logger.error(f"Erreur réglage consigne : {e}")

    def action(self, args: list[str]) -> None:
        """Point d'entrée pour toutes les actions en ligne de commande."""
        cmd = args[0]
        logger.info(f"Action reçue : {cmd}")

        if cmd == "dhw":
            self._set_dhw_temperature(args)

        elif cmd in {str(l) for l in (Level.STANDBY, Level.NORMAL, Level.COMFORT, Level.ECO, Level.DHW)}:
            self._set_mode(int(cmd))

        else:
            self._set_temperature(cmd)


# ---------------------------------------------------------------------------
# Point d'entrée
# ---------------------------------------------------------------------------
def main() -> None:
    try:
        bridge = VicareBridge()
        if len(sys.argv) > 1:
            bridge.action(sys.argv[1:])
        else:
            bridge.sync()
    except Exception as e:
        logger.error(f"Erreur fatale : {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()

Code : Tout sélectionner

# ViCare API credentials
VICARE_USER=xxxxxx@gmail.com
VICARE_PASS=xxxxxxxx
VICARE_CLIENT_ID=xxxxxxxxxx

# Chemins et URLs (optionnel — valeurs par défaut si absent)
TOKEN_FILE=/home/pi/domoticz/scripts/python/token.json
DOMOTICZ_URL=http://192.168.1.45:9090

Code : Tout sélectionner

return {
    on = {
        devices = {
            600, -- IDX Chauffage
            601, -- IDX Eau Chaude
            599  -- IDX Sélecteur
        }
    },
    execute = function(domoticz, device)
        if (device.id == 600) then
            -- Envoi de la température de chauffage
            domoticz.executeShellCommand('python3 /home/pi/domoticz/scripts/python/vicare.py ' .. device.setPoint)
            domoticz.log('Commande Chauffage envoyée : ' .. device.setPoint, domoticz.LOG_INFO)
            
        elseif (device.id == 601) then
            -- Envoi de la température Eau Chaude
            domoticz.executeShellCommand('python3 /home/pi/domoticz/scripts/python/vicare.py dhw ' .. device.setPoint)
            domoticz.log('Commande Eau Chaude envoyée : ' .. device.setPoint, domoticz.LOG_INFO)
            
        elseif (device.id == 599) then
            -- Envoi du niveau du sélecteur (0, 10, 20, 30)
            domoticz.executeShellCommand('python3 /home/pi/domoticz/scripts/python/vicare.py ' .. device.level)
            domoticz.log('Commande Mode envoyée, niveau : ' .. device.level, domoticz.LOG_INFO)
        end
    end
}