first commit

This commit is contained in:
2021-05-24 10:23:05 +02:00
commit 370afa3374

419
plugin.py Executable file
View File

@@ -0,0 +1,419 @@
# Heatzy python plugin for Domoticz
#
# Author: fjumelle
#
"""
<plugin key="Heatzy_FJU" name="Heatzy Pilote" author="fjumelle" version="1.0.2" wikilink="" externallink="">
<description>
<h2>Heatzy Pilote</h2><br/>
Implementation of Heatzy Pilote as a Domoticz Plugin.<br/>
</description>
<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="Mode4" label="'Off=Normal' bug?" width="200px">
<options>
<option label="No" value="0" default="true"/>
<option label="Yes" value="1"/>
</options>
</param>
<param field="Mode5" label="Polling interval (s)" width="40px" required="true" default="60"/>
<param field="Mode6" label="Logging Level" width="200px">
<options>
<option label="Normal" value="0" default="true"/>
<option label="Verbose" value="1"/>
</options>
</param>
</params>
</plugin>
"""
import Domoticz
import requests
import json
import time
import urllib.parse as parse
import urllib.request as request
from datetime import datetime, timedelta
HEATZY_MODE = {
'停止': 'OFF',
'解冻': 'FROSTFREE',
'经济': 'ECONOMY',
'舒适': 'NORMAL'
}
HEATZY_MODE_NAME = {
'OFF': 'Off',
'FROSTFREE': 'Hors Gel',
'ECONOMY': 'Eco',
'NORMAL': 'Confort'
}
HEATZY_MODE_VALUE = {
'OFF': 0,
'FROSTFREE': 10,
'ECONOMY': 20,
'NORMAL': 30
}
HEATZY_MODE_VALUE_INV = {v: k for k, v in HEATZY_MODE_VALUE.items()}
DEFAULT_POOLING = 60
class deviceparam:
def __init__(self, unit, nvalue, svalue):
self.unit = unit
self.nvalue = nvalue
self.svalue = svalue
class BasePlugin:
debug = False
token = ""
token_expire_at = 0
did = ""
mode = 0
nextupdate = datetime.now()
bug = False
pooling = 30
pooling_steps = 1
pooling_current_step = 1
max_retry = 6
retry = max_retry
def __init__(self):
return
def onStart(self):
import math
# setup the appropriate logging level
debuglevel = int(Parameters["Mode6"])
if debuglevel != 0:
self.debug = True
Domoticz.Debugging(debuglevel)
DumpConfigToLog()
else:
self.debug = False
Domoticz.Debugging(0)
# Polling interval = X sec
try:
pooling = int(Parameters["Mode5"])
except:
pooling = DEFAULT_POOLING
self.pooling_steps = math.ceil(pooling/30)
self.pooling = pooling // self.pooling_steps
Domoticz.Heartbeat(self.pooling)
# create the child devices if these do not exist yet
devicecreated = []
if 1 not in Devices:
Domoticz.Device(Name="Control", Unit=1, TypeName="Switch", Image=9, Used=1).Create()
devicecreated.append(deviceparam(1, 0, "Off")) # default is Off
if 2 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()
devicecreated.append(deviceparam(2, 0, "30")) # default is confort mode
# 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.getToken(Parameters["Username"], Parameters["Password"])
# Get Devide Id
self.did = self.getDevideId(self.token)
# Get mode
self.mode = self.getMode()
def onCommand(self, Unit, Command, Level, Hue):
if Unit == 1:
self.mode = self.onOff(Command)
elif Unit == 2:
self.mode = self.setMode(Level)
def onHeartbeat(self):
if self.pooling_current_step >= self.pooling_steps:
Domoticz.Debug("Retry counter:{}".format(self.retry))
if self.retry < 0:
Domoticz.Log("No connection to Heatzy API ==> Device disabled")
self.pooling_current_step = 1
return
self.mode = self.getMode()
# 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.setMode(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.pooling_current_step = 1
else:
self.pooling_current_step = self.pooling_current_step + 1
def getToken(self, user, password):
import time
need_to_get_token = False
if self.token == "" or self.token_expire_at == "":
need_to_get_token = True
Domoticz.Status("Heatzy Token unknown, need to call Heatzy API.")
elif (float(self.token_expire_at)-time.time()) < 24*60*60:
#Token will expire in less than 1 day
need_to_get_token = True
Domoticz.Status("Heatzy Token expired, need to call Heatzy API.")
if need_to_get_token and self.retry>=0:
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json',
'X-Gizwits-Application-Id': 'c70a66ff039d41b4a220e198b0fcc8b3',
}
data = '{ "username": "'+user+'", "password": "'+password+'", "lang": "en" }'
try:
time.sleep(0.5)
url = 'https://euapi.gizwits.com/app/login'
response = requests.post(url, headers=headers, data=data, timeout=3).json()
except Exception as exc:
Domoticz.Error("Cannot open connection to Heatzy API to get the token: " + str(exc))
#Domoticz.Error("URL: " + str(url))
#Domoticz.Error("Headers: " + str(headers))
Domoticz.Error("Data: " + str(data))
#Decrease retry
self.retry = self.retry - 1
return "", ""
Domoticz.Debug("Get Token Response: " + str(response))
if 'token' in response:
self.token = response['token']
self.token_expire_at = response['expire_at']
Domoticz.Status("Token from Heatzy API: " + self.token)
#Reset retry counter
self.retry = self.max_retry
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']
Domoticz.Error("Cannot get Heatzy Token: {} ({})\n{}".format(error_message, error_code, response))
self.token = ""
self.token_expire_at = 0
self.did = ""
#Decrease retry
self.retry = self.retry - 1
return self.token, self.token_expire_at
def getDevideId(self, token):
if token == "" or self.retry<0:
return ""
if self.did == "":
Domoticz.Status("Heatzy Devide Id unknown, need to call Heatzy API.")
headers = {
'Accept': 'application/json',
'X-Gizwits-User-token': token,
'X-Gizwits-Application-Id': 'c70a66ff039d41b4a220e198b0fcc8b3',
}
params = (('limit', '20'), ('skip', '0'),)
url = 'https://euapi.gizwits.com/app/bindings'
try:
response = requests.get(url, headers=headers, params=params, timeout=3).json()
except Exception as exc:
Domoticz.Error("Cannot open connection to Heatzy API to get the device id: " + str(exc))
#Domoticz.Error("URL: " + str(url))
#Domoticz.Error("Headers: " + str(headers))
#Domoticz.Error("Params: " + str(params))
return ""
Domoticz.Debug("Get Device Id Response:" + str(response))
if 'devices' in response and 'did' in response['devices'][0]:
self.did = response['devices'][0]['did']
Domoticz.Status("Devide Id from Heatzy API: " + self.did)
else:
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("Cannot get Heatzy Devide Id: {} ({})\n{}".format(error_message, error_code, response))
return self.did
def getMode(self):
mode = ""
response = ""
self.token, self.token_expire_at = self.getToken(Parameters["Username"], Parameters["Password"])
self.did = self.getDevideId(self.token)
if self.retry<0:
return ""
headers = {
'Accept': 'application/json',
'X-Gizwits-Application-Id': 'c70a66ff039d41b4a220e198b0fcc8b3',
}
url = 'https://euapi.gizwits.com/app/devdata/{}/latest'.format(self.did)
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 ""
Domoticz.Debug("Get Mode Response:" + str(response))
if 'attr' in response and 'mode' in response['attr']:
mode = HEATZY_MODE[response['attr']['mode']]
Domoticz.Debug("Current Heatzy Mode: {}".format(HEATZY_MODE_NAME[mode]))
#Reset retry counter
self.retry = self.max_retry
if Devices[2].nValue != HEATZY_MODE_VALUE[mode]:
Domoticz.Status("New Heatzy Mode: {}".format(HEATZY_MODE_NAME[mode]))
Devices[2].Update(nValue=HEATZY_MODE_VALUE[mode], sValue=str(HEATZY_MODE_VALUE[mode]), TimedOut = 0)
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)
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
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("Cannot get Heatzy Mode: {} ({})\n{}\nToken: {}\nDeviceId: {}".format(error_message, error_code, response, self.token, self.did))
if error_code == 9004:
#Invalid token
self.token = ""
self.did = ""
return ""
return mode
def setMode(self, mode):
if Devices[2].nValue != mode:
mode_str = {
HEATZY_MODE_VALUE['NORMAL']: '[1,1,0]',
HEATZY_MODE_VALUE['ECONOMY']: '[1,1,1]',
HEATZY_MODE_VALUE['FROSTFREE']: '[1,1,2]',
HEATZY_MODE_VALUE['OFF']: '[1,1,3]'
}
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json',
'X-Gizwits-User-token': self.token,
'X-Gizwits-Application-Id': 'c70a66ff039d41b4a220e198b0fcc8b3',
}
data = '{"raw": '+mode_str[mode]+'}'
url = 'https://euapi.gizwits.com/app/control/{}'.format(self.did)
try:
response = requests.post(url, headers=headers, data=data, timeout=3).json()
except Exception as exc:
Domoticz.Error("Cannot open connection to Heatzy API to set the mode: " + str(exc))
#Domoticz.Error("URL: " + str(url))
#Domoticz.Error("Headers: " + str(headers))
Domoticz.Error("Data: " + str(data))
return self.mode
Domoticz.Debug("Set Mode Response:" + str(response))
if response != None:
self.mode = HEATZY_MODE_VALUE_INV[mode]
Devices[2].Update(nValue=int(mode), sValue=str(mode))
Domoticz.Status("New Heatzy Mode: {}".format(HEATZY_MODE_NAME[self.mode]))
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)
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)
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']
Domoticz.Error("Cannot set Heatzy Mode: {} ({})\n{}\nToken: {}\nDeviceId: {}".format(error_message, error_code, response, self.token, self.did))
if error_code == 9004:
#Invalid token
self.token = ""
self.did = ""
return self.mode
return self.mode
def onOff(self, command):
if Devices[1].sValue != command:
if command == "On":
self.mode = self.setMode(HEATZY_MODE_VALUE['NORMAL'])
else:
if not self.bug:
self.mode = self.setMode(HEATZY_MODE_VALUE['OFF'])
else:
#Because of issue with the equipment (Off do not work...)
self.mode = self.setMode(HEATZY_MODE_VALUE['FROSTFREE'])
return self.mode
global _plugin
_plugin = BasePlugin()
def onStart():
global _plugin
_plugin.onStart()
def onCommand(Unit, Command, Level, Hue):
global _plugin
_plugin.onCommand(Unit, Command, Level, Hue)
def onHeartbeat():
global _plugin
_plugin.onHeartbeat()
# Generic helper functions
def DumpConfigToLog():
for x in Parameters:
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))
return