Closed riddik14 closed 1 year ago
# based on https://github.com/jezcooke/haier_appdaemon/blob/main/checkappliance.py import hassapi as hass import requests import json import codecs from datetime import datetime, timedelta, timezone encryption_key: "EEDcwokGsbwHqvjJ" appliance_host: "192.168.1.129" appliance_entity = "sensor.candy_wifi" # The name of the entity to use/create in Home Assistant (value with '_stats' appended) appliance_entity_binary_sensor = "binary_sensor.candy_wifi" # The name of the entity to use/create in Home Assistant (value with '_stats' appended) status_root = "statusLavatrice" # The root level JSON element returned by 'http-read' power_attribute = "MachMd" # The name of the JSON attribute that containes the power on/off state. stats_root = "statusCounters" # The root level JSON element returned by 'http-getStatistics' polling_interval = 60 # How frequently check for the latest status. request_timeout = 30 # Request timeout should be less than the polling interval max_retry_before_unavailable = 3 PROGRAM_STATES = { 0: "SPENTA", 4: "DELICATI 59 MINUTI", 7: "RAPIDO", 8: "COTONE PERFETTO", 12: "CICLO TEST", 40: "IGIENE PLUS", 72: "SPORT PLUS 39 MINUTI", 104: "AUTO-PULIZIA", 135: "MISTI & COLORATI 59 MINUTI", 136: "SPECIAL 39 MINUTI" } POWER = { 1: "STOP", 2: "LAVAGGIO IN CORSO", 3: "PAUSA", 4: "PARTENZA RITARDATA SELEZIONATA", 5: "PARTENZA RITARDATA", 6: "ERRORE", 7: "FINITO", 8: "SCONOSCIUTO" } PORTA = { 10: "aperta", 6: "Chiusa", 7: "Bloccata" } FASE_LAVAGGIO = { 1: "prelavaggio", 2: "lavaggio", 3: "risciacquo", 4: "ultimo risciacquo", 5: "fine", 6: "asciugatura", 7: "ERRORE", 8: "vapore", 9: "centrifuga notturna", 10: "centrifuga" } LIVELLI_SPORCO = { 1: "Poco", 2: "Normale", 3: "Molto" } n_risciacquo = { } class CandyWashingMachine(hass.Hass): def initialize(self): self.retry = 0 self.previous_end = None self.run_every(self.check_appliance, "now", polling_interval) self.encryption_key = self.args["encryption_key"] self.log(f"encryption_key: {self.encryption_key!r}") self.appliance_host = self.args["appliance_host"] def check_appliance(self, kwargs): try: status = self.get_status() power = int(status[status_root][power_attribute]) state_power = POWER.get(power, "UNKNOWN") attributes = status[status_root] self.set_state(appliance_entity, state=state_power, attributes=attributes) #{"friendly_name": "Candy Lavatrice", "icon": "mdi:washing-machine" }) self.retry = 0 remaining_minutes = int(attributes["RemTime"]) / 60 + int(attributes["DelVal"]) now_rounded = datetime.now(timezone.utc).replace(second=0, microsecond=0) + timedelta(minutes=1) end = now_rounded + timedelta(minutes=remaining_minutes) if self.previous_end is not None: if abs(end - self.previous_end) <= timedelta(minutes=1): end = max(end, self.previous_end) self.previous_end = end entity_id = appliance_entity + "_termine_programma" self.set_state( entity_id, state=remaining_minutes, attributes={"friendly_name": "Candy Fine ","icon": "mdi:av-timer", "unit_of_measurement": "minuti"}, ) except Exception as e: self.log(f"error when getting status: {e}") self.retry += 1 if self.retry > max_retry_before_unavailable: self.set_state(appliance_entity, state="SPENTA") entity_id = appliance_entity + "_termine_programma" self.set_state( entity_id, state="unavailable", attributes={"friendly_name": "Candy Lavatrice", "device_class": "timestamp", "icon": "mdi:timer-off-outline"}, ) previous_end = None return #################################### try: status = self.get_status() opta = int(status[status_root]["Opt5"]) optb = int(status[status_root]["Opt6"]) optc = int(status[status_root]["Opt7"]) if opta == 1: state_risc = "X 1" elif optb == 1: state_risc = "X 2" elif optc == 1: state_risc = "X 3" else: state_risc = "OFF" entity_id = f"{appliance_entity}_risciacquo" attributes = {"friendly_name": "Risciacquo", "icon": "mdi:water"} self.set_state(entity_id, state=state_risc, attributes=attributes) self.retry = 0 except Exception as e: # Gestione specifica dell'eccezione, se necessario pass #livello sporco try: statussp = self.get_status() sporco = int(statussp[status_root]["SLevel"]) statesp = LIVELLI_SPORCO.get(sporco, "ESCLUSO") entity_id = appliance_entity + "_livello_sporco" self.set_state( entity_id, state=statesp, attributes={ "friendly_name": "Livello di Sporco","icon": "mdi:car-brake-fluid-level",}, ) except: pass #stato lavaggio try: status = self.get_status() macchine = int(status[status_root]["PrPh"]) statemd = stati_lavatrice.get(macchine, "inattiva") entity_id = appliance_entity + "_stato_lavatrice" self.set_state( entity_id, state=statemd, attributes={ "friendly_name": "Candy Stato", "icon": "mdi:washing-machine", }, ) self.retry = 0 except: pass #porta try: status = self.get_status() ntcd = int(status[status_root]["NtcD"]) stato_porta = PORTA.get(ntcd, "UNKNOW") entity_id = appliance_entity + "_ntcd" self.set_state(entity_id, state=stato_porta, attributes = {"friendly_name": "Porta", "icon":"mdi:door"}) #self.retry = 0 except: stato_porta = "UNKNOWN" ############### #cicli conta try: entity_id = appliance_entity + "_stats" stats = self.get_stats()[stats_root] total = 0 for (key, value) in stats.items(): if key.startswith("Program"): total += int(value) self.set_state(entity_id, state=total, attributes={ "friendly_name": "Candy Cicli totali", "icon": "mdi:washing-machine"},) except: pass ###################################### #programma try: status = self.get_status() programm = int(status[status_root]["PrCode"]) rem = int(status[status_root]["RemTime"]) // 60 state_program = PROGRAM_STATES.get(programm, "UNKNOWN") if programm == 7: state_program += f" {rem} MINUTI" entity_id = appliance_entity + "_programma" self.set_state(entity_id, state=state_program, attributes = {"friendly_name": "Programma", "icon":"mdi:format-list-bulleted-type"}) #self.retry = 0 except: state_program = "UNKNOWN" #errori try: # Ottieni lo stato del sistema status = self.get_status() error_code = int(status[status_root]["Err"]) # Assegna una stringa di errore in base al codice di errore state_error = "E{}".format(error_code) if error_code in range(1, 22) else "---" # Imposta lo stato di errore sull'entità entity_id = appliance_entity + "_errore" self.set_state(entity_id, state=state_error, attributes = {"friendly_name": "ERRORI", "icon":"mdi:alert-circle-outline"}) except: pass ############### #temp impostata try: status = self.get_status() temp = int(status[status_root]["Temp"]) entity_id = appliance_entity + "_temp" self.set_state(entity_id, state=temp, attributes = {"friendly_name": "Temperatura impostata", "unit_of_measurement": "°C"}) # self.retry = 0 except: pass ############### #temp interna try: status = self.get_status() temp1 = int(status[status_root]["NtcW"]) / 10 entity_id = appliance_entity + "_temp_interna" self.set_state(entity_id, state=temp1, attributes = {"friendly_name": "Temperatura attuale", "unit_of_measurement": "°C"}) # self.retry = 0 except: pass ############### ############### #counter try: statusmv = self.get_stats() mov = int(statusmv[stats_root]["CounterMV"]) mov = 0 entity_id = appliance_entity + "_countmove" self.set_state(entity_id, state=mov, attributes = {"friendly_name": "Conta Movimenti", "icon":"mdi:vibrate"}) # self.retry = 0 except: pass ############### #riempimento try: statusfill = self.get_status() fill = int(statusfill[status_root]["FillR"]) fill = 0 entity_id = appliance_entity + "_riempimento" self.set_state(entity_id, state=fill, attributes = {"friendly_name": "Percentuale riempimento", "icon":"mdi:waves-arrow-up", "unit_of_measurement": "%"}) # self.retry = 0 except: pass #centrifuga impostata try: statusrpm = self.get_status() rpm = int(statusrpm[status_root]["SpinSp"]) * 100 entity_id = appliance_entity + "_centrifuga" self.set_state(entity_id, state=rpm, attributes = {"friendly_name": "Giri Centrifuga", "unit_of_measurement": "rpm", "icon":"mdi:sync"}) self.retry = 0 except: pass ############### ############### #motore try: status = self.get_status() value = int(status[status_root]["motS"]) // 10 entity_id = appliance_entity + "_motore" self.set_state(entity_id, state=value, attributes = {"friendly_name": "Giri Motore", "unit_of_measurement": "rpm", "icon": "mdi:engine"}) # self.retry = 0 except: pass #prelavaggio opt1 try: status = self.get_status() prelavax = int(status[status_root]["Opt1"]) if prelavax == 1: state = "on" else: state = "off" entity_id = appliance_entity_binary_sensor + "_opt1" self.set_state(entity_id, state=state, attributes = {"friendly_name": "Prelavaggio", "icon":"mdi:hand-wash"}) # self.retry = 0 except: pass ############### #igiene opt2 try: status = self.get_status() igieneplus = int(status[status_root]["Opt2"]) if igieneplus == 1: state = "on" else: state = "off" entity_id = appliance_entity_binary_sensor + "_opt2" self.set_state(entity_id, state=state, attributes = {"friendly_name": "Igiene +", "icon":"mdi:hospital-box"}) # self.retry = 0 except: pass ############### #antipiega opt3 try: status = self.get_status() valueantip = int(status[status_root]["Opt3"]) if valueantip == 1: state = "on" else: state = "off" entity_id = appliance_entity_binary_sensor + "_opt3" self.set_state(entity_id, state=state, attributes = {"friendly_name": "Antipiega", "icon":"mdi:tshirt-v"}) # self.retry = 0 except: pass ############### #buonanotte opt4 try: status = self.get_status() value = int(status[status_root]["Opt4"]) if value == 1: state = "on" else: state = "off" entity_id = appliance_entity_binary_sensor + "_opt4" self.set_state(entity_id, state=state, attributes = {"friendly_name": "Buonanotte", "icon":"mdi:weather-night"}) # self.retry = 0 except: pass #acquaplus opt8 try: status = self.get_status() valueacq = int(status[status_root]["Opt8"]) if valueacq == 1: state = "on" else: state = "off" entity_id = appliance_entity_binary_sensor + "_opt8" self.set_state(entity_id, state=state, attributes = {"friendly_name": "Acquaplus", "icon":"mdi:water-plus"}) # self.retry = 0 except: pass ############### #option9 opt9 try: status = self.get_status() value = int(status[status_root]["Opt9"]) if value == 1: state = "on" else: state = "off" entity_id = appliance_entity_binary_sensor + "_opt9" self.set_state( entity_id, state=state, attributes = {"friendly_name": "Opzione 9 sconosciuta"}) # self.retry = 0 except: pass #vapore try: status = self.get_status() valuevap = int(status[status_root]["Steam"]) states = ["ESCLUSO", "Basso", "Medio Basso", "Medio", "Medio Alto", "Massimo"] state = states[valuevap] if 0 <= valuevap <= 5 else "ESCLUSO" entity_id = f"{appliance_entity}_vapore" self.set_state( entity_id, state=state, attributes={"friendly_name": "Vapore", "icon": "mdi:cloud-outline"} ) except Exception as e: # Logging dell'errore per una diagnostica futura logger.exception("Errore durante la gestione del vapore: %s", str(e)) ###################################@@@@@@@@@@@@@@@@@@@@@############################## #controllo remoto try: status = self.get_status() valueremoto = int(status[status_root]["WiFiStatus"]) stato_wifi = { 1: "on", 0: "off", } controllorem = stato_wifi.get(valueremoto, "off") entity_id = appliance_entity_binary_sensor + "_wifi" self.set_state( entity_id, state=controllorem, attributes={ "friendly_name": "Controllo Remoto", "WiFiStatus": valueremoto, "icon": "mdi:wifi-cog", }, ) self.retry = 0 except: pass #filtro try: statsf = self.get_stats()[stats_root] totalef = sum(int(value) for key, value in statsf.items() if key.startswith("Program")) filtro_lav = 100 - totalef if filtro_lav < 1: statefiltros = "Da Pulire" elif filtro_lav < 70: statefiltros = "Medio Sporco" elif filtro_lav < 40: statefiltros = "Sporco" else: statefiltros = "Pulito" entity_id = f"{appliance_entity}_filtro" self.set_state( entity_id, state=statefiltros, attributes={ "friendly_name": "Filtro", "Intasamento": filtro_lav, "icon": "mdi:air-filter" } ) except Exception as e: # Logging dell'errore per una diagnostica futura logger.exception("Errore durante la gestione del filtro: %s", str(e)) #filtro calcare try: statsfc = self.get_stats()[stats_root] totalefc = sum(int(value) for key, value in statsfc.items() if key.startswith("Program")) filtro_lav_c = 105 - totalefc if filtro_lav_c < 1: statefiltroc = "Da Pulire" elif filtro_lav_c < 70: statefiltroc = "Medio Sporco" elif filtro_lav_c < 40: statefiltroc = "Sporco" else: statefiltroc = "Pulito" entity_id = f"{appliance_entity}_filtro_calcare" self.set_state( entity_id, state=statefiltroc, attributes={ "friendly_name": "Filtro Calcare", "Livello": filtro_lav_c, "icon": "mdi:air-filter" } ) self.retry = 0 except Exception as e: # Logging dell'errore per una diagnostica futura logger.exception("Errore durante la gestione del filtro del calcare: %s", str(e)) #################################### #diagnosi try: status = self.get_status() diagnostics_test_on = int(status[status_root]["DisTestOn"]) diagnostics_test_result = int(status[status_root]["DisTestRes"]) if diagnostics_test_on == 1: state_diagnostics = "In Corso" elif diagnostics_test_result == 1: state_diagnostics = "OK" elif diagnostics_test_result == 2: state_diagnostics = "ERRORE" else: state_diagnostics = "---" entity_id = f"{appliance_entity}_diagnostics" self.set_state( entity_id, state=state_diagnostics, attributes={ "friendly_name": "Diagnosi", "icon": "mdi:medical-bag" } ) self.retry = 0 except Exception as e: # Logging l'errore per una diagnostica futura logger.exception(f"Errore durante la gestione della diagnosi: {e}") pass ################################## ################################## def get_status(self): return self.get_data("read") def get_stats(self): self.get_data("prepareStatistics") return self.get_data("getStatistics") def get_data(self, command): res = requests.get( "http://" + self.appliance_host + "/http-" + command + ".json?encrypted=1", timeout=request_timeout, ) return json.loads(self.decrypt(codecs.decode(res.text, "hex"), self.encryption_key)) def decrypt(self, cipher_text, key): decrypted = "" for i in range(len(cipher_text)): decrypted += chr(cipher_text[i] ^ ord(key[i % len(key)])) return decrypted