J’ai mon chargeur de voiture qui est sous SmartLife/Tuya. Le plugin étant à l’abandon, je me suis fait un script pour intégrer et commander mon chargeur via Jeedom. J’ai passé une dizaine d’heure là dessus avec pas mal de galères surtout pour l’access token.
Alors si ça peut aider d’autres personnes…
Pré-requis avoir le plugin MQTT (ou autre) pour avoir un broker Mosquitto sur son Jeedom.
- Créer un compte sur https://iot.tuya.com
- Il faut ensuite y créer un projet : Cloud →Project Management → Create Cloud Project, par exemple dans mons cas « ChargeurVE » en catégorie Smart Home.
- Lier ton compte Tuya / Smart Life Project Management → Open project → Devices → Link App Account et scanner le QR Code avec l’application Smart Life sur téléphone.
- Ajouter les Services API de bases : IoT Core (par défaut normalement)
- Récupérer le Device ID de l’objet dans Cloud → API Explorer.
Ensuite le reste se passe sur le Pi. Mon script n’est pas parfait, la méthodologie peut-être pas non plus, je ne suis pas un dev, mais ça marche !
A vous de le personnaliser, ou de le développer plus si besoin.
- J’ai installé les librairies nécessaires :
pip install paho-mqtt requests
- Dans mon répertoire home (mais on peut le faire dans un répertoire Jeedom comme celui du plugin Script/data) j’ai créé un fichier Python : tuya_to_mqtt.py
Avec ce contenu (je ne vais pas tout commenter mais si besoin je préciserais) :
import time
import hashlib
import hmac
import requests
import threading
import json
import paho.mqtt.client as mqtt
import logging
from logging.handlers import RotatingFileHandler
# Configuration Tuya
CLIENT_ID = "ton_client_id_sur_projet_tuya"
SECRET = "ton_secret_sur_projet_tuya"
BASEURL = "https://openapi.tuyaeu.com" # pour l'Europe
DEVICE_ID = "ton_id_du_device"
# Configuration MQTT
MQTT_BROKER = "localhost"
MQTT_PORT = 1883
MQTT_USER = "ton_user"
MQTT_PASS = "ton_pass"
MQTT_TOPIC = "tuya/chargeurVE" # nom du topic mqtt principal de l'objet
MQTT_TOPIC_CMD = "tuya/chargeurVE/cmd" # nom du topic des commandes
MQTT_TOPIC_ERR = "tuya/chargeurVE/erreur" # nom du topic des erreurs
# Logger avec rotation automatique
file_handler = RotatingFileHandler(
"/var/log/tuya_to_mqtt.log",
maxBytes=5*1024*1024, # 5 Mo max par fichier
backupCount=7 # Garder 7 fichiers de rotation
)
logging.basicConfig(
level=logging.INFO,
format="[%(asctime)s.%(msecs)03d][%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
handlers=[file_handler, logging.StreamHandler()]
)
EmptyBodyEncoded = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
last_properties = {} # Stocke les dernières valeurs pour éviter les publications inutiles
update_thread_active = False
def on_connect(client, userdata, flags, rc, properties=None):
logging.info(f"📡 Connecté au broker MQTT (code {rc})")
client.subscribe(MQTT_TOPIC_CMD)
def on_message(client, userdata, msg):
try:
payload = json.loads(msg.payload.decode().strip())
code, value = payload.get("code"), payload.get("value")
if code and value is not None:
send_command_to_device(token, code, value)
else:
logging.warning(f"⚠️ Commande MQTT mal formée : {payload}")
except Exception as e:
logging.error(f"❌ Erreur traitement MQTT cmd : {e}")
mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
mqtt_client.username_pw_set(MQTT_USER, MQTT_PASS)
mqtt_client.on_connect = on_connect
mqtt_client.on_message = on_message
mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
mqtt_client.loop_start()
time.sleep(0.2)
# Fonction pour générer la signature Tuya
def generate_signature(url, access_token=None):
timestamp = str(int(time.time() * 1000))
string_to_sign = f"{CLIENT_ID}{timestamp}GET\n{EmptyBodyEncoded}\n\n{url}"
if access_token:
string_to_sign = f"{CLIENT_ID}{access_token}{timestamp}GET\n{EmptyBodyEncoded}\n\n{url}"
sign = hmac.new(SECRET.encode(), string_to_sign.encode(), hashlib.sha256).hexdigest().upper()
return timestamp, sign
# Obtenir un token d'accès
def get_access_token():
url = "/v1.0/token?grant_type=1"
retry_count = 3 # Nombre de tentatives avant alerte
for attempt in range(retry_count):
timestamp, sign = generate_signature(url)
headers = {
"sign_method": "HMAC-SHA256",
"client_id": CLIENT_ID,
"t": timestamp,
"Content-Type": "application/json",
"sign": sign,
}
try:
logging.info(f"🔄 Récupération du token (tentative {attempt + 1}/{retry_count})...")
response = requests.get(BASEURL + url, headers=headers, timeout=5)
data = response.json()
if data.get("success"):
result = data.get("result", {})
token = result.get("access_token")
if token:
logging.info(f"✅ Nouveau token : {token}")
return token
else:
error_msg = "❌ Réponse valide mais token manquant"
logging.critical(error_msg)
mqtt_client.publish(MQTT_TOPIC_ERR, json.dumps({"erreur": error_msg}))
return None
else:
error_msg = f"Erreur token inconnue : {data.get('msg', 'Non spécifié')} (code {error_code})"
logging.warning(f"⚠️ {error_msg}")
mqtt_client.publish(MQTT_TOPIC_ERR, json.dumps({"erreur": error_msg}))
return None
except requests.RequestException as e:
logging.warning(f"❌ Erreur réseau (tentative token {attempt + 1}/{retry_count}) : {e}")
if attempt < retry_count - 1:
time.sleep(5 * (attempt + 1))
continue
error_msg = f"❌ Erreur réseau persistante token : {e}"
mqtt_client.publish(MQTT_TOPIC_ERR, json.dumps({"erreur": error_msg}))
logging.critical(error_msg)
return None
# Échec après toutes les tentatives
error_msg = f"❌ Échec token après {retry_count} tentatives. Vérifier la console Tuya IOT."
logging.critical(error_msg)
mqtt_client.publish(f"{MQTT_TOPIC_BASE}/erreur", json.dumps({"erreur": error_msg}))
return None
# Obtenir les propriétés d'un appareil
def get_device_properties(access_token):
url = f"/v2.0/cloud/thing/{DEVICE_ID}/shadow/properties"
timestamp, sign = generate_signature(url, access_token)
headers = {
"sign_method": "HMAC-SHA256",
"client_id": CLIENT_ID,
"t": timestamp,
"mode": "cors",
"Content-Type": "application/json",
"sign": sign,
"access_token": access_token
}
try:
response = requests.get(BASEURL + url, headers=headers, timeout=5)
return response.json()
except requests.RequestException as e:
logging.error(f"❌ Erreur réseau sur récupération des propriétés : {e}")
return {}
# Fonction pour envoyer une commande
def send_command_to_device(access_token, code, value):
url = f"/v1.0/devices/{DEVICE_ID}/commands"
body = json.dumps({"commands": [{"code": code, "value": value}]}).encode("utf-8")
body_hash = hashlib.sha256(body).hexdigest()
timestamp = str(int(time.time() * 1000))
string_to_sign = f"{CLIENT_ID}{access_token}{timestamp}POST\n{body_hash}\n\n{url}"
sign = hmac.new(SECRET.encode(), string_to_sign.encode(), hashlib.sha256).hexdigest().upper()
headers = {
"sign_method": "HMAC-SHA256",
"client_id": CLIENT_ID,
"t": timestamp,
"Content-Type": "application/json",
"sign": sign,
"access_token": access_token
}
try:
response = requests.post(BASEURL + url, headers=headers, data=body, timeout=5).json()
if response.get("code") == 1010:
logging.info("🔄 Token expiré, renouvellement, commande en attente...")
token = get_access_token()
if token:
logging.info(f"✅ Nouveau token : {token}, renvoi de la commande...")
return send_command_to_device(token, code, value)
error_message = {"erreur": "❌ Impossible de récupérer un nouveau token, commande annulée"}
mqtt_client.publish(MQTT_TOPIC_ERR, json.dumps(error_message))
logging.error(f"[{DEVICE_ID}] {error_message['erreur']}")
return None
logging.info(f"📤 Commande envoyée ({code} = {value}) -> Réponse : {response}")
# Rafraîchir les propriétés après commande sans interrompre la boucle principale
time.sleep(1)
threading.Thread(target=update_device_state, args=(access_token,)).start()
return access_token
except requests.RequestException as e:
error_message = {"erreur": f"❌ Erreur réseau lors de l'envoi de la commande : {e}"}
mqtt_client.publish(MQTT_TOPIC_ERR, json.dumps(error_message))
logging.error(f"[{DEVICE_ID}] {error_message['erreur']}")
return None
# Fonction pour rafraîchir l'état et publier dans MQTT
def update_device_state(access_token):
global update_thread_active
if update_thread_active:
return
update_thread_active = True
try:
device_properties = get_device_properties(access_token)
if device_properties.get("success"):
properties = device_properties.get("result", {}).get("properties", [])
logging.info(f"🔄 Mise à jour après commande : [\n" + ",\n".join(json.dumps(item) for item in properties) + "\n]")
mqtt_client.publish(MQTT_TOPIC, json.dumps(properties))
else:
logging.warning("⚠️ Impossible de rafraîchir après la commande.")
finally:
update_thread_active = False
# Exécution principale
if __name__ == "__main__":
token = get_access_token()
token_creation_time = time.time() # Timestamp du token
interval = 60 # relever les propriétés toutes les minutes par défaut
while True:
if token is None:
logging.warning("Impossible d'obtenir un token. Nouvelle essai dans 3 minutes...") # ne pas surcharger l'API si un problème critique de token
time.sleep(180)
token = get_access_token()
if token:
token_creation_time = time.time()
continue
token_age = time.time() - token_creation_time # Vérifie si le token a dépassé sa durée de vie
if token_age >= 7200:
logging.info(f"⏰ Token a {token_age:.0f}s, considéré comme expiré (>= 7200s), renouvellement...")
token = get_access_token()
if token:
token_creation_time = time.time()
continue
device_properties = get_device_properties(token)
if device_properties.get("msg") == "token invalid":
logging.info("⚠️ Token invalide détecté, renouvellement...")
token = get_access_token()
if token:
token_creation_time = time.time()
continue
if device_properties.get("success"):
properties = device_properties.get("result", {}).get("properties", [])
properties_dict = {item["code"]: item["value"] for item in properties}
if properties_dict != last_properties:
logging.info("📡 Nouvelles propriétés récupérées : [\n" + ",\n".join(json.dumps(item) for item in properties) + "\n]")
mqtt_client.publish(MQTT_TOPIC, json.dumps(properties))
last_properties = properties_dict # Mise en mémoire des valeurs
else:
logging.info("⚡ Aucune modification, pas d'envoi MQTT.")
else:
logging.error(f"⚠️ Erreur de récupération des propriétés : {device_properties.get('msg')}")
logging.info(f"⏳ Attente de {interval} secondes")
time.sleep(interval)
Il ne vous reste plus qu’à lancer le script et voir si ça marche :
python3 tuya_to_mqtt.py
Si tout est ok, il devrait y avoir une remontée dans le plugin mqtt de Jeedom soit en auto-découverte soit en l’ajoutant à la main avec le bon topic racine : MQTT_TOPIC et en cochant Activer l’analyse des valeurs pour récupérer les commandes simplement.
Pour le reste c’est comme tout équipement mqtt, vous pouvez l’afficher sur Jeedom mais pour l’instant vous ne recevez l’info que si vous avez lancé le script, on va donc automatiser ça.
Avec un Raspberry Pi sous Debian, on va commencer par créer le fichier log et lui donner les droits qu’il faut
sudo touch /var/log/tuya_to_mqtt.log
sudo chown ton_username:ton_username/var/log/tuya_to_mqtt.log
sudo chmod 644 /var/log/tuya_to_mqtt.log
et on va utiliser systemd pour exécuter le script en tant que service.
sudo nano /etc/systemd/system/tuya_to_mqtt.service
on écrit
[Unit]
Description=Tuya to MQTT bridge
After=network.target
[Service]
ExecStart=/usr/bin/python3 /chemin/du/script/tuya_to_mqtt.py
Restart=always
User=pi
WorkingDirectory=/répertoire/du/script
StandardOutput=append:/var/log/tuya_to_mqtt.log
StandardError=append:/var/log/tuya_to_mqtt.log
[Install]
WantedBy=multi-user.target
on active
sudo systemctl daemon-reload
sudo systemctl enable tuya_to_mqtt
sudo systemctl start tuya_to_mqtt
On teste
sudo systemctl status tuya_to_mqtt
Si tout est bon, le statut doit être « active (running) »
Voilà le script est en place et se lancera à chaque reboot ou si plantage !
Dans la partie Jeedom, pour l’équipement on peut envoyer des commandes, pour cela on lui ajoute une commande de type action vers le topic MQTT_TOPIC_CMD :
Ici c’est un slider pour modifier l’ampérage du chargeur, le code est :
{code: "charge_cur_set", "value": #slider#}
attention à bien retirer les " autour de code, #slider# récupère la valeur. Il ne reste plus qu’à bouger le slider et vérifier le résultat ! 
On peut aussi optimiser la boucle de répétition While pour ne pas surcharger l’API et pour que le script interroge plus ou moins souvent suivant l’utilisation de l’objet.
Voilà, je crois avoir tout dit ! Si besoin je répondrais (à mon petit niveau).