tuya_project/tuya-control/devices/lights/all_lights_effects_toggle.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Script pour basculer l'état de plusieurs lumières d'ambiance simultanément:
- Saber du bureau dino
- Ringlight
- Chinalight
- Windowled

À utiliser avec StreamDeck
"""

import sys
import os
import time
import logging
import tinytuya

# Ajouter le répertoire parent au path pour les imports
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
from common.utils import report_for_streamdeck
from common import config

# Obtenir le logger
logger = logging.getLogger("TuyaControl")

# ID et paramètres des appareils
DEVICES = [
    {
        "id": "820180048cce4e2aecc7", 
        "name": "Saber (Bureau Dino)", 
        "switch": "7",  # Code pour la prise USB (saber)
        "ip": "192.168.1.160",
        "key": "_rgNtA^X&Vp)A|^l",
        "version": 3.3
    },
    {
        "id": "bfadb5f27d0fe9a778ixwc", 
        "name": "Ringlight", 
        "switch": "20",
        "ip": "192.168.1.70",
        "key": "dOo5*OR3ixjP?OxO",
        "version": 3.5
    },
    {
        "id": "bff3ca58c60d8b18d4x3dn", 
        "name": "Chinalight", 
        "switch": "20",
        "ip": "192.168.1.122",
        "key": "Nt6*~S4<2x9;?Q+-",
        "version": 3.5
    },
    {
        "id": "bf10d8832c8447c512jedz", 
        "name": "Windowled", 
        "switch": "20",
        "ip": "192.168.1.8",
        "key": "e#Z$#dvgmRo}nS8^",
        "version": 3.5
    }
]

def connect_device_direct(device_info):
    """
    Se connecte directement à un appareil avec son IP sans passer par la découverte
    """
    try:
        # Créer l'objet appareil
        device = tinytuya.Device(
            dev_id=device_info["id"],
            address=device_info["ip"],
            local_key=device_info["key"],
            version=device_info["version"]
        )
        
        # Configurer le timeout
        device.set_socketTimeout(config.TIMEOUT)
        
        # Vérifier la connexion (optionnel)
        status = device.status()
        if 'Error' in status:
            logger.error(f"Erreur de connexion à {device_info['name']} ({device_info['ip']}): {status['Error']}")
            return None
            
        logger.info(f"Connexion réussie à {device_info['name']} ({device_info['ip']})")
        return device
        
    except Exception as e:
        logger.error(f"Erreur lors de la connexion à {device_info['name']} ({device_info['ip']}): {e}")
        return None

def main():
    """Fonction principale pour basculer l'état des lumières d'ambiance simultanément"""
    # Connexion aux appareils
    connected_devices = []
    
    for device_info in DEVICES:
        try:
            # Ajouter un petit délai entre chaque connexion pour éviter les collisions de socket
            time.sleep(0.5)
            # Connexion directe avec l'IP
            device = connect_device_direct(device_info)
            if device:
                connected_devices.append({
                    "device": device,
                    "name": device_info["name"],
                    "switch": device_info["switch"]
                })
        except Exception as e:
            logger.error(f"Erreur de connexion à {device_info['name']}: {e}")
    
    if not connected_devices:
        report_for_streamdeck(False, "Ambiance")
        return
        
    try:
        # Déterminer l'état actuel pour décider de la nouvelle action
        lights_on = False
        
        for device_data in connected_devices:
            try:
                status = device_data["device"].status()
                if status.get('dps', {}).get(device_data["switch"], False):
                    lights_on = True
                    break
            except Exception as e:
                logger.error(f"Erreur lors de la récupération du statut de {device_data['name']}: {e}")
        
        # Définir le nouvel état (inverse de l'état actuel)
        new_state = not lights_on
        
        # Appliquer l'action sur tous les appareils
        success = True
        for device_data in connected_devices:
            try:
                result = device_data["device"].set_value(device_data["switch"], new_state)
                if result is None or "Error" in str(result):
                    success = False
                    logger.error(f"Erreur lors du changement d'état de {device_data['name']}")
                # Ajouter un petit délai entre chaque commande
                time.sleep(0.2)
            except Exception as e:
                success = False
                logger.error(f"Erreur lors du changement d'état de {device_data['name']}: {e}")
        
        # Rapport pour StreamDeck
        report_for_streamdeck(success, "Ambiance", new_state)
        
    finally:
        # Fermer explicitement toutes les connexions
        for device_data in connected_devices:
            try:
                device_data["device"].set_socketRetryLimit(0)
                device_data["device"].set_socketTimeout(1)
                if hasattr(device_data["device"], "_connection") and device_data["device"]._connection:
                    device_data["device"]._connection.close()
            except:
                pass

if __name__ == "__main__":
    main()