# 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
![dependabot[bot] avatar](https://avatars.githubusercontent.com/in/29110?v=4 "dependabot[bot]")
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent