import tinytuya import requests import time import math import numpy as np from scipy import stats from datetime import datetime def get_value_safely(data, key): try: # Tente d'obtenir la valeur associée à la clé, si elle existe value = data['dps'].get(key, None) # Si la valeur est None, renvoie 0 return value if value is not None else 0 except Exception as e: # En cas d'erreur, imprime un message d'erreur et renvoie 0 print(f"Erreur : {e}") return 0 # Obtenir la date et l'heure actuelles maintenant = datetime.now() # Formater la date et l'heure format_date_heure = maintenant.strftime("%Y-%m-%d %H:%M:%S") # Imprimer la date et l'heure print("Date et Heure actuelles :", format_date_heure) # Connect to Device #"apiKey": "hgkvdyrstyhqudku3xdu", #"apiSecret": "704a1bad394a4efe805f46fb04c46916", #"apiRegion": "eu-w", #"apiDeviceID": "bf4d021ef10d39ea17gpwo" response = "" previous_sens = "" watt_1_p = 0 watt_2_p = 0 PUISSANCE_DELESTAGE = 750 # Watts MAX_VALUES = 5 # Initialisation de la liste des dernières valeurs last_values = [] # Initialisation de la somme des valeurs sum_values = 0 # Initialisation du compteur pour suivre le nombre de valeurs ajoutées counter = 0 def remove_outliers(values): if len(values) < 2: return values z_scores = stats.zscore(values) abs_z_scores = np.abs(z_scores) filtered_entries = (abs_z_scores < 3) return values[filtered_entries] while True: d = tinytuya.OutletDevice( dev_id='bf9109f5435d9bb3c6avlr', address='192.168.1.14', # Or set to 'Auto' to auto-discover IP address local_key='/Pht&M$(/nt@t}qV', version=3.4) # Get Status data = d.status() print('set_status() result %r' % data) #data = json.loads(data) # for key, value in data['dps'].items(): # print(f'{key.ljust(5)}: {str(value).ljust(10)}') # Extraire les valeurs des champs spécifiques #total = data['dps'].get('115', None) / 10 # Utilisation de la fonction get_value_safely pour remplacer les lignes existantes total_1 = get_value_safely(data, '1') total_2 = get_value_safely(data, '2') volt_1 = get_value_safely(data, '112') / 10 or 0 freq_1 = get_value_safely(data, '111') / 100 or 0 current_1 = get_value_safely(data, '113') or 0 factor_1 = get_value_safely(data, '110') / 100 or 0 prod_1 = get_value_safely(data, '107') or 0 watt_1 = get_value_safely(data, '101') / 10 or 0 watt_2 = get_value_safely(data, '105') / 10 or 0 prod_2 = get_value_safely(data, '108') or 0 current_2 = get_value_safely(data, '114') or 0 sens_1 = get_value_safely(data, '102') sens_2 = get_value_safely(data, '104') value_106 = get_value_safely(data, '106') or 0 value_109 = get_value_safely(data, '109') or 0 # Afficher les valeurs print("=================================") print("1 kWh :", total_1 ) print("2 kWh :", total_2 ) # print("101 total :", total_1) print("101 Watt 1 :", watt_1) print("105 Watt 2 :", watt_2) print("102 Sens 1 :", sens_1) print("104 Sens 2 :", sens_2) print("106 conso jour 1:", value_106) print("107 prod jour 1 :", prod_1) print("108 conso jour 2:", prod_2) print("109 prod jour 2:", value_109) print("110 factor :", factor_1) print("111 freq :", freq_1) print("112 Volt :", volt_1) print("113 Current 1 :", current_1) print("114 Current 2 :", current_2) #print("115 :", value_115 / 10) print("=================================") # Turn On #d.turn_on() #if (sens_1 == "FORWARD" and watt_1 > watt_2 - 100): # sens_1 = "NORMAL" if (watt_1 == 0 and watt_2 == 0): time.sleep(10) continue if (watt_1 == watt_1_p and watt_2 == watt_2_p): time.sleep(10) continue watt_1_p = watt_1 watt_2_p = watt_2 if (sens_1 != previous_sens): previous_sens = sens_1 time.sleep(15) continue previous_sens = sens_1 # id domoticz 1181 # print('1121|0|' + str(value_105) + ";0;" + str(value_115) + ";0;" + str(value_108) + ";" + str(value_118)) #if sens_1 == "REVERSE": # watt_1 = 0 # 1087 Solaire production solaire = watt_2 if watt_2 > 0 else 0 if watt_1 < 0 and solaire < - watt_1 + 90: time.sleep(15) continue value = (watt_1 if sens_1 == "FORWARD" else - watt_1) # Tuya Device 2 url = "http://192.168.1.3:81/json.htm?type=command¶m=udevice&idx=1182&nvalue=0&svalue=" + str(total_1) + ";0;" + str(total_2) + ";0;" + str(watt_1) + ";" + str(solaire) #USAGE1;USAGE2;RETURN1;RETURN2;CONS;PROD if watt_1 or watt_2: print(url) try: response = requests.get(url) print(response) except requests.RequestException as e: # Gérez les autres erreurs liées aux requêtes ici print(f"Erreur de requête : {e}") time.sleep(1) #value=moyenne idx = "1087" url = "http://192.168.1.3:81/json.htm?type=command¶m=udevice&idx=" + idx + "&svalue=" + str(solaire) + ";0&nvalue=0"; #response = requests.get(url) #print(response) time.sleep(1) # 1123 Consommation_Apparente last_values.append(value) sum_values += value counter += 1 # Vérifier si la liste des dernières valeurs dépasse éléments if len(last_values) > MAX_VALUES: # Si oui, retirer la première valeur de la liste et ajuster la somme sum_values -= last_values.pop(0) # Calculer la moyenne des dernières valeurs print(f"Valeurs non filtrées : {last_values}") # Convertir en numpy array pour le filtrage #last_values_np = np.array(last_values) # Enlever les valeurs aberrantes #last_values_np = remove_outliers(last_values_np) # Convertir le résultat filtré en liste et mettre à jour last_values #last_values = list(last_values_np) print(f"Valeurs filtrées (sans aberrantes): {last_values}") moyenne = sum_values / min(counter, MAX_VALUES) #value=moyenne #watt_1=moyenne # Utiliser la moyenne comme vous le souhaitez (par exemple, l'afficher) print("Moyenne des dernières valeurs :", moyenne) if 1 == 1: #(sens_1 == previous_sens): #if value < 0 and (abs(value) > watt_2): # value = -value #if value < 0 and abs(value) > solaire - 100: # time.sleep(3) # continue idx = "1123" url = "http://192.168.1.3:81/json.htm?type=command¶m=udevice&idx=" + idx + "&svalue=" + str(value) + ";0&nvalue=0"; try: response = requests.get(url) print(response) except requests.RequestException as e: # Gérez les autres erreurs liées aux requêtes ici print(f"Erreur de requête : {e}") time.sleep(1) # 1138 Mesure Courant idx = "1138" url = "http://192.168.1.3:81/json.htm?type=command¶m=udevice&idx=" + idx + "&svalue=" + str(watt_2) + ";0&nvalue=0"; try: response = requests.get(url) print(response) except requests.RequestException as e: # Gérez les autres erreurs liées aux requêtes ici print(f"Erreur de requête : {e}") time.sleep(1) # 1115 CONSOMMATION_GENERALE idx = "1115" url = "http://192.168.1.3:81/json.htm?type=command¶m=udevice&idx=" + idx + "&svalue=" + str(watt_1) + ";0&nvalue=0"; try: #response = requests.get(url) print(response) except requests.RequestException as e: # Gérez les autres erreurs liées aux requêtes ici print(f"Erreur de requête : {e}") previous_sens = sens_1 time.sleep(5)