fjumelle/domoticz-Heatzy
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

Move to DomoticsEx API and support multiple devices in the same plugin

Browse Source
This commit is contained in:
François JUMELLE
2024-12-08 23:51:05 +01:00
parent 8942775fbc
commit 76a4169308
1 changed files with 162 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

304
plugin.py
View File

@@ -4,7 +4,7 @@
#
#pylint: disable=line-too-long,broad-exception-caught,possibly-used-before-assignment
"""
<plugin key="Heatzy_FJU" name="Heatzy Pilote" author="fjumelle" version="1.2.0" wikilink="" externallink="">
<plugin key="HeatzyEx" name="Heatzy Pilote Ex" author="fjumelle" version="2.0.0" wikilink="" externallink="">
<description>
<h2>Heatzy Pilote</h2><br/>
Implementation of Heatzy Pilote as a Domoticz Plugin.<br/>
@@ -12,7 +12,6 @@
<params>
<param field="Username" label="Heatzy Username" width="200px" required="true" default=""/>
<param field="Password" label="Heatzy Password" width="200px" required="true" default="" password="true"/>
<param field="Mode3" label="Device name" width="200px" required="true" default=""/>
<param field="Mode4" label="'Off=Normal' bug?" width="200px">
<options>
<option label="No" value="0" default="true"/>
@@ -34,7 +33,7 @@ import time
from datetime import datetime
import requests
import Domoticz # type: ignore
import DomoticzEx as Domoticz # type: ignore
if None is not None: #Fake statement to remove warning on global Domoticz variables #NOSONAR
Parameters = Parameters # type: ignore #NOSONAR #pylint: disable=undefined-variable,self-assigning-variable
@@ -75,8 +74,7 @@ class BasePlugin:
debug = False
token = ""
token_expire_at = 0
did = ""
mode = 0
did = {}
nextupdate = datetime.now()
bug = False
pooling = 30
@@ -84,7 +82,6 @@ class BasePlugin:
pooling_current_step = 1
max_retry = 6
retry = max_retry
last_update = 0
def __init__(self):
return
@@ -110,38 +107,44 @@ class BasePlugin:
self.pooling = pooling // self.pooling_steps
Domoticz.Heartbeat(self.pooling)
# create the child devices if these do not exist yet
if 1 not in Devices:
Domoticz.Device(Name="Control", Unit=1, TypeName="Switch", Image=9, Used=1).Create()
if 2 not in Devices:
# Get Token
self.token, self.token_expire_at = self.get_token(Parameters["Username"], Parameters["Password"])
# Get Devide Id
self.did = self.get_heatzy_devices()
# Create the child devices if these do not exist yet
if "Control" not in Devices:
for did in self.did:
unit = self.did[did]["unit"]
Domoticz.Unit(Name=f"Heatzy {did} - Control", DeviceID="Control", Unit=unit, TypeName="Switch", Image=9, Used=1).Create()
if "Mode" not in Devices:
options = {"LevelActions": "||",
"LevelNames": HEATZY_MODE_NAME['OFF'] + "|" + HEATZY_MODE_NAME['FROSTFREE'] + "|" + HEATZY_MODE_NAME['ECONOMY'] + "|" + HEATZY_MODE_NAME['NORMAL'],
"LevelOffHidden": "false", #Bug with off mode...
#"LevelOffHidden": "true",t
"SelectorStyle": "0"}
Domoticz.Device(Name="Mode", Unit=2, TypeName="Selector Switch", Switchtype=18, Image=15,
Options=options, Used=1).Create()
for did in self.did:
unit = self.did[did]["unit"]
Domoticz.Unit(Name=f"Heatzy {did} - Mode", DeviceID="Mode", Unit=unit,
TypeName="Selector Switch", Switchtype=18, Image=15,
Options=options, Used=1).Create()
# Bug Off = Normal?
if str(Parameters["Mode4"]) != "0":
Domoticz.Log("Heatzy plugin configured to support the bug Off=Confort. Then when switching to Off, Heatzy will switch to Frost Free.")
self.bug = True
# Get Token
self.token, self.token_expire_at = self.get_token(Parameters["Username"], Parameters["Password"])
# Get Devide Id
self.did = self.get_device_id(self.token, Parameters["Mode3"])
# Get mode
self.mode = self.get_mode()
self.get_mode()
def on_command(self, unit, command, level, hue): #pylint: disable=unused-argument
def on_command(self, DeviceID, Unit, Command, Level, Color): #pylint: disable=unused-argument
"""Send a command"""
if unit == 1:
self.mode = self.on_off(command)
elif unit == 2:
self.mode = self.set_mode(level)
if DeviceID == "Control":
self.on_off(Unit, Command)
elif DeviceID == "Mode":
self.set_mode(Unit, Level)
def on_heartbeat(self):
"""Time to heartbeat :)"""
@@ -157,18 +160,7 @@ class BasePlugin:
self.did = ""
return
self.mode = self.get_mode()
# If mode = OFF and bug, then mode = FROSTFREE
if self.bug and self.mode == 'OFF':
Domoticz.Log("Switch to FROSTFREE because of the OFF bug...")
self.mode = self.set_mode(HEATZY_MODE_VALUE['FROSTFREE'])
# check if need to refresh device so that they do not turn red in GUI
#now = datetime.now()
#if self.nextupdate <= now:
# self.nextupdate = now + timedelta(minutes=int(Settings["SensorTimeout"]))
# Devices[2].Update(nValue=Devices[2].nValue, sValue=Devices[2].sValue)
self.get_mode()
self.pooling_current_step = 1
else:
@@ -226,17 +218,17 @@ class BasePlugin:
return self.token, self.token_expire_at
def get_device_id(self, token, alias):
"""Get the device id from the token and the device name, using the Heatzy API"""
if token == "" or self.retry<0:
def get_heatzy_devices(self):
"""Get the device id from the token, using the Heatzy API"""
if self.token == "" or self.retry<0:
return ""
if self.did == "":
if len(self.did) == 0:
Domoticz.Status("Heatzy Devide Id unknown, need to call Heatzy API.")
headers = {
'Accept': 'application/json',
'X-Gizwits-User-token': token,
'X-Gizwits-User-token': self.token,
'X-Gizwits-Application-Id': 'c70a66ff039d41b4a220e198b0fcc8b3',
}
params = (('limit', '20'), ('skip', '0'),)
@@ -252,20 +244,16 @@ class BasePlugin:
Domoticz.Debug("Get Device Id Response:" + str(response))
found = False
if 'devices' in response:
devices = response['devices']
unit = 1
for device in devices:
if "dev_alias" in device and "did" in device and device["dev_alias"].lower() == alias.lower():
found = True
self.did = device['did']
Domoticz.Status("Devide Id from Heatzy API: " + self.did)
if not found:
self.did = ""
error_code = "Unknown" if 'error_code' not in response else response['error_code']
error_message = "Unknown" if 'error_message' not in response else response['error_message']
Domoticz.Error(f"Cannot get Heatzy Devide Id: {error_message} ({error_code})\n{response}")
if "dev_alias" in device and "did" in device:
alias = device['dev_alias']
did = device['did']
self.did[alias] = {"did":did, "unit":unit, "last_update":0}
Domoticz.Status(f"Devide Id from Heatzy API: {alias} - {did}")
unit = unit + 1
return self.did
@@ -275,7 +263,7 @@ class BasePlugin:
response = ""
self.token, self.token_expire_at = self.get_token(Parameters["Username"], Parameters["Password"])
self.did = self.get_device_id(self.token, Parameters["Mode3"])
self.did = self.get_heatzy_devices()
if self.retry<0:
return ""
@@ -285,76 +273,92 @@ class BasePlugin:
'X-Gizwits-User-token': self.token,
'X-Gizwits-Application-Id': 'c70a66ff039d41b4a220e198b0fcc8b3',
}
url = f"https://euapi.gizwits.com/app/devdata/{self.did}/latest"
try:
response = requests.get(url, headers=headers, timeout=3).json()
except Exception as exc:
#Decrease retry
self.retry = self.retry - 1
if self.retry < self.max_retry//2:
Domoticz.Error("Cannot open connection to Heatzy API to get the mode: " + str(exc))
#Domoticz.Error("URL: " + str(url))
#Domoticz.Error("Headers: " + str(headers))
if 'response' in locals() and response != "":
Domoticz.Error("Response: " + str(response))
return ""
for device_name in self.did:
device = self.did[device_name]
did = device["did"]
unit = device["unit"]
Domoticz.Debug("Get Mode Response:" + str(response))
url = f"https://euapi.gizwits.com/app/devdata/{did}/latest"
try:
response = requests.get(url, headers=headers, timeout=3).json()
except Exception as exc:
#Decrease retry
self.retry = self.retry - 1
#Last Update
if 'updated_at' in response:
if self.last_update != response['updated_at']:
self.last_update = response['updated_at']
else:
obsolete_min = int((time.time() - self.last_update)//60)
if obsolete_min >= 30 and obsolete_min%60 == 0: #more than 30 min and every hour
Domoticz.Error(f"Heatzy Mode obsolete! {obsolete_min} min")
if self.retry < self.max_retry//2:
Domoticz.Error("Cannot open connection to Heatzy API to get the mode: " + str(exc))
#Domoticz.Error("URL: " + str(url))
#Domoticz.Error("Headers: " + str(headers))
if 'response' in locals() and response != "":
Domoticz.Error("Response: " + str(response))
return ""
if 'attr' in response and 'mode' in response['attr']:
mode = HEATZY_MODE[response['attr']['mode']]
Domoticz.Debug(f"Current Heatzy Mode: {HEATZY_MODE_NAME[mode]}")
Domoticz.Debug("Get Mode Response:" + str(response))
#Reset retry counter
self.retry = self.max_retry
#Last Update
if 'updated_at' in response:
if device["last_update"] != response['updated_at']:
device["last_update"] = response['updated_at']
else:
obsolete_min = int((time.time() - device["last_update"])//60)
if obsolete_min >= 30 and obsolete_min%60 == 0: #more than 30 min and every hour
Domoticz.Error(f"Heatzy Mode obsolete! {obsolete_min} min")
return ""
if Devices[2].nValue != HEATZY_MODE_VALUE[mode]:
Domoticz.Status(f"New Heatzy Mode: {HEATZY_MODE_NAME[mode]}")
Devices[2].Update(nValue=HEATZY_MODE_VALUE[mode], sValue=str(HEATZY_MODE_VALUE[mode]), TimedOut = 0)
if 'attr' in response and 'mode' in response['attr']:
mode = HEATZY_MODE[response['attr']['mode']]
Domoticz.Debug(f"Current Heatzy Mode: {HEATZY_MODE_NAME[mode]} ({self.get_name(unit)})")
if not self.bug:
if mode == 'OFF' and Devices[1].nValue != 0:
Devices[1].Update(nValue=0, sValue="Off", TimedOut = 0)
elif mode != 'OFF' and Devices[1].nValue == 0:
Devices[1].Update(nValue=1, sValue="On", TimedOut = 0)
#Reset retry counter
self.retry = self.max_retry
if Devices["Mode"].Units[unit].nValue != HEATZY_MODE_VALUE[mode]:
Domoticz.Status(f"New Heatzy Mode: {HEATZY_MODE_NAME[mode]} ({self.get_name(unit)})")
Devices["Mode"].Units[unit].nValue = HEATZY_MODE_VALUE[mode]
Devices["Mode"].Units[unit].sValue = str(HEATZY_MODE_VALUE[mode])
Devices["Mode"].Units[unit].Update()
if not self.bug:
if mode == 'OFF' and Devices["Control"].Units[unit].nValue != 0:
Devices["Control"].Units[unit].nValue = 0
Devices["Control"].Units[unit].sValue = "Off"
elif mode != 'OFF' and Devices["Control"].Units[unit].nValue == 0:
Devices["Control"].Units[unit].nValue = 1
Devices["Control"].Units[unit].sValue = "On"
else:
if mode in ('OFF', 'FROSTFREE') and Devices["Control"].Units[unit].nValue != 0:
Devices["Control"].Units[unit].nValue = 0
Devices["Control"].Units[unit].sValue = "Off"
elif mode not in ('OFF', 'FROSTFREE') and Devices["Control"].Units[unit].nValue == 0:
Devices["Control"].Units[unit].nValue = 1
Devices["Control"].Units[unit].sValue = "On"
Devices["Control"].Units[unit].Update()
else:
if mode in ('OFF', 'FROSTFREE') and Devices[1].nValue != 0:
Devices[1].Update(nValue=0, sValue="Off", TimedOut = 0)
elif mode not in ('OFF', 'FROSTFREE') and Devices[1].nValue == 0:
Devices[1].Update(nValue=1, sValue="On", TimedOut = 0)
else:
#Decrease retry
self.retry = self.retry - 1
#Decrease retry
self.retry = self.retry - 1
error_code = "Unknown" if 'error_code' not in response else response['error_code']
error_message = "Unknown" if 'error_message' not in response else response['error_message']
Domoticz.Error(f"Cannot get Heatzy Mode: {error_message} ({error_code})\n{response}\nToken: {self.token}\nDeviceId: {self.did}")
if error_code == 9004:
#Invalid token
self.token = ""
self.did = ""
elif 'attr' in response and len(response["attr"]) == 0:
#attr is empty...
Domoticz.Status("We force a setMode to try to get the correct mode at the next try...")
self.set_mode(HEATZY_MODE_VALUE['FROSTFREE'])
return ""
error_code = "Unknown" if 'error_code' not in response else response['error_code']
error_message = "Unknown" if 'error_message' not in response else response['error_message']
Domoticz.Error(f"Cannot get Heatzy Mode: {error_message} ({error_code})\n{response}\nToken: {self.token}\nDeviceId: {self.did}")
if error_code == 9004:
#Invalid token
self.token = ""
self.did = {}
elif 'attr' in response and len(response["attr"]) == 0:
#attr is empty...
Domoticz.Status("We force a setMode to try to get the correct mode at the next try...")
self.set_mode(unit, HEATZY_MODE_VALUE['FROSTFREE'])
return ""
return mode
# If mode = OFF and bug, then mode = FROSTFREE
if self.bug and mode == 'OFF':
Domoticz.Log(f"Switch to FROSTFREE because of the OFF bug...({device})")
self.set_mode(unit, HEATZY_MODE_VALUE['FROSTFREE'])
def set_mode(self, mode):
def set_mode(self, unit, mode):
"Set the device mode using the Heatzy API"
if Devices[2].nValue != mode:
if Devices["Mode"].Units[unit].nValue != mode:
mode_str = {
HEATZY_MODE_VALUE['NORMAL']: 0, #'[1,1,0]',
HEATZY_MODE_VALUE['ECONOMY']: 1, #'[1,1,1]',
@@ -369,7 +373,9 @@ class BasePlugin:
}
#data = '{"raw": '+mode_str[mode]+'}'
data = '{"attrs": {"mode":' + str(mode_str[mode]) + '}}'
url = f"https://euapi.gizwits.com/app/control/{self.did}"
device_name = self.get_name(unit)
did = self.did[device_name]["did"]
url = f"https://euapi.gizwits.com/app/control/{did}"
try:
response = requests.post(url, headers=headers, data=data, timeout=3).json()
except Exception as exc:
@@ -377,24 +383,31 @@ class BasePlugin:
#Domoticz.Error("URL: " + str(url))
#Domoticz.Error("Headers: " + str(headers))
Domoticz.Error("Data: " + str(data))
return self.mode
return
Domoticz.Debug("Set Mode Response:" + str(response))
if response is not None:
self.mode = HEATZY_MODE_VALUE_INV[mode]
Devices[2].Update(nValue=int(mode), sValue=str(mode))
Domoticz.Status(f"New Heatzy Mode: {HEATZY_MODE_NAME[self.mode]}")
mode_str = HEATZY_MODE_VALUE_INV[mode]
Devices["Mode"].Units[unit].nValue = int(mode)
Devices["Mode"].Units[unit].sValue = str(mode)
Devices["Mode"].Units[unit].Update()
Domoticz.Status(f"New Heatzy Mode: {HEATZY_MODE_NAME[mode_str]} ({self.get_name(unit)})")
if not self.bug:
if self.mode == 'OFF' and Devices[1].nValue != 0:
Devices[1].Update(nValue=0, sValue="Off", TimedOut = 0)
elif self.mode != 'OFF' and Devices[1].nValue == 0:
Devices[1].Update(nValue=1, sValue="On", TimedOut = 0)
if mode_str == 'OFF' and Devices["Control"].Units[unit].nValue != 0:
Devices["Control"].Units[unit].nValue = 0
Devices["Control"].Units[unit].sValue = "Off"
elif mode_str != 'OFF' and Devices["Control"].Units[unit].nValue == 0:
Devices["Control"].Units[unit].nValue = 1
Devices["Control"].Units[unit].sValue = "On"
else:
if self.mode in ('OFF', 'FROSTFREE') and Devices[1].nValue != 0:
Devices[1].Update(nValue=0, sValue="Off", TimedOut = 0)
elif self.mode not in ('OFF', 'FROSTFREE') and Devices[1].nValue == 0:
Devices[1].Update(nValue=1, sValue="On", TimedOut = 0)
if mode_str in ('OFF', 'FROSTFREE') and Devices["Control"].Units[unit].nValue != 0:
Devices["Control"].Units[unit].nValue = 0
Devices["Control"].Units[unit].sValue = "Off"
elif mode_str not in ('OFF', 'FROSTFREE') and Devices["Control"].Units[unit].nValue == 0:
Devices["Control"].Units[unit].nValue = 1
Devices["Control"].Units[unit].sValue = "On"
Devices["Control"].Units[unit].Update()
else:
error_code = "Unknown" if 'error_code' not in response else response['error_code']
error_message = "Unknown" if 'error_message' not in response else response['error_message']
@@ -402,27 +415,30 @@ class BasePlugin:
if error_code == 9004:
#Invalid token
self.token = ""
self.did = ""
return self.mode
self.did = {}
return
#Device is correctly running ==> we reset the retry counter
self.retry = self.max_retry
return self.mode
def on_off(self, command):
def on_off(self, unit, command):
"""Toggle device on/off"""
if Devices[1].sValue != command:
if Devices["Control"].Units[unit].sValue != command:
if command == "On":
self.mode = self.set_mode(HEATZY_MODE_VALUE['NORMAL'])
self.set_mode(unit, HEATZY_MODE_VALUE['NORMAL'])
else:
if not self.bug:
self.mode = self.set_mode(HEATZY_MODE_VALUE['OFF'])
self.set_mode(unit, HEATZY_MODE_VALUE['OFF'])
else:
#Because of issue with the equipment (Off do not work...)
self.mode = self.set_mode(HEATZY_MODE_VALUE['FROSTFREE'])
self.set_mode(unit, HEATZY_MODE_VALUE['FROSTFREE'])
return self.mode
def get_name(self, unit):
"""get device name form unit"""
for device_name in self.did:
if self.did[device_name]["unit"] == unit:
return device_name
return "Not found"
_plugin = BasePlugin()
@@ -430,9 +446,9 @@ def onStart(): #NOSONAR #pylint: disable=invalid-name
"""OnStart"""
_plugin.on_start()
def onCommand(Unit, Command, Level, Hue): #NOSONAR #pylint: disable=invalid-name
def onCommand(DeviceID, Unit, Command, Level, Color): #NOSONAR #pylint: disable=invalid-name
"""OnCommand"""
_plugin.on_command(Unit, Command, Level, Hue)
_plugin.on_command(DeviceID, Unit, Command, Level, Color)
def onHeartbeat(): #NOSONAR #pylint: disable=invalid-name
"""onHeartbeat"""
@@ -445,11 +461,15 @@ def dump_config_to_log():
if Parameters[x] != "":
Domoticz.Debug( "'" + x + "':'" + str(Parameters[x]) + "'")
Domoticz.Debug("Device count: " + str(len(Devices)))
for x in Devices:
Domoticz.Debug("Device: " + str(x) + " - " + str(Devices[x]))
Domoticz.Debug("Device ID: '" + str(Devices[x].ID) + "'")
Domoticz.Debug("Device Name: '" + Devices[x].Name + "'")
Domoticz.Debug("Device nValue: " + str(Devices[x].nValue))
Domoticz.Debug("Device sValue: '" + Devices[x].sValue + "'")
Domoticz.Debug("Device LastLevel: " + str(Devices[x].LastLevel))
for DeviceName in Devices:
Device = Devices[DeviceName]
Domoticz.Debug("Device ID: '" + str(Device.DeviceID) + "'")
Domoticz.Debug("--->Unit Count: '" + str(len(Device.Units)) + "'")
for UnitNo in Device.Units:
Unit = Device.Units[UnitNo]
Domoticz.Debug("--->Unit: " + str(UnitNo))
Domoticz.Debug("--->Unit Name: '" + Unit.Name + "'")
Domoticz.Debug("--->Unit nValue: " + str(Unit.nValue))
Domoticz.Debug("--->Unit sValue: '" + Unit.sValue + "'")
Domoticz.Debug("--->Unit LastLevel: " + str(Unit.LastLevel))
return