diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5d2da6d --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +Domoticz.py +parameters.properties +run.py +__pycache__/ diff --git a/plugin.py b/plugin.py index 1318fe3..8d8222f 100755 --- a/plugin.py +++ b/plugin.py @@ -2,6 +2,7 @@ # # Author: fjumelle # +#pylint: disable=line-too-long,broad-exception-caught,possibly-used-before-assignment """ @@ -11,6 +12,7 @@ + """ -import Domoticz -import requests -import json +import math import time -import urllib.parse as parse -import urllib.request as request -from datetime import datetime, timedelta +from datetime import datetime + +import requests +import 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 + Images = Images # type: ignore #NOSONAR #pylint: disable=undefined-variable,self-assigning-variable + Devices = Devices # type: ignore #NOSONAR #pylint: disable=undefined-variable,self-assigning-variable HEATZY_MODE = { '停止': 'OFF', @@ -60,14 +66,8 @@ 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: + """Class for plugin""" debug = False token = "" token_expire_at = 0 @@ -84,15 +84,14 @@ class BasePlugin: def __init__(self): return - def onStart(self): - import math - + def on_start(self): + """At statup""" # setup the appropriate logging level debuglevel = int(Parameters["Mode6"]) if debuglevel != 0: self.debug = True Domoticz.Debugging(debuglevel) - DumpConfigToLog() + dump_config_to_log() else: self.debug = False Domoticz.Debugging(0) @@ -100,26 +99,23 @@ class BasePlugin: # Polling interval = X sec try: pooling = int(Parameters["Mode5"]) - except: + except Exception: 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": "||", + 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 + Options=options, Used=1).Create() # Bug Off = Normal? if str(Parameters["Mode4"]) != "0": @@ -127,36 +123,38 @@ class BasePlugin: self.bug = True # Get Token - self.token, self.token_expire_at = self.getToken(Parameters["Username"], Parameters["Password"]) + self.token, self.token_expire_at = self.get_token(Parameters["Username"], Parameters["Password"]) # Get Devide Id - self.did = self.getDevideId(self.token) + self.did = self.get_device_id(self.token, Parameters["Mode3"]) # 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) + self.mode = self.get_mode() - def onHeartbeat(self): + def on_command(self, unit, command, level, hue): #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) + + def on_heartbeat(self): + """Time to heartbeat :)""" if self.pooling_current_step >= self.pooling_steps: - Domoticz.Debug("Retry counter:{}".format(self.retry)) + Domoticz.Debug(f"Retry counter:{self.retry}") if self.retry < 0: Domoticz.Status("No connection to Heatzy API ==> Device disabled for 15 minutes") self.pooling_current_step = - 15 * 60 // self.pooling + self.pooling_steps self.retry = self.max_retry return - self.mode = self.getMode() + 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.setMode(HEATZY_MODE_VALUE['FROSTFREE']) - + 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: @@ -166,10 +164,9 @@ class BasePlugin: self.pooling_current_step = 1 else: self.pooling_current_step = self.pooling_current_step + 1 - - def getToken(self, user, password): - import time - + + def get_token(self, user, password): + """Get token using the Heatzy API""" need_to_get_token = False if self.token == "" or self.token_expire_at == "": @@ -179,7 +176,7 @@ class BasePlugin: #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', @@ -211,19 +208,20 @@ class BasePlugin: 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)) + Domoticz.Error(f"Cannot get Heatzy Token: {error_message} ({error_code})\n{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): + 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: return "" - + if self.did == "": Domoticz.Status("Heatzy Devide Id unknown, need to call Heatzy API.") @@ -242,41 +240,48 @@ class BasePlugin: #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: + found = False + if 'devices' in response: + devices = response['devices'] + 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("Cannot get Heatzy Devide Id: {} ({})\n{}".format(error_message, error_code, response)) + Domoticz.Error(f"Cannot get Heatzy Devide Id: {error_message} ({error_code})\n{response}") return self.did - def getMode(self): + def get_mode(self): + "Get the device mode using the Heatzy API" mode = "" response = "" - - self.token, self.token_expire_at = self.getToken(Parameters["Username"], Parameters["Password"]) - self.did = self.getDevideId(self.token) + + self.token, self.token_expire_at = self.get_token(Parameters["Username"], Parameters["Password"]) + self.did = self.get_device_id(self.token, Parameters["Mode3"]) 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) + 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)) @@ -289,13 +294,13 @@ class BasePlugin: 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])) + Domoticz.Debug(f"Current Heatzy Mode: {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])) + 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 not self.bug: @@ -314,16 +319,17 @@ class BasePlugin: 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)) + 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 = "" return "" - + return mode - def setMode(self, mode): + def set_mode(self, mode): + "Set the device mode using the Heatzy API" if Devices[2].nValue != mode: mode_str = { HEATZY_MODE_VALUE['NORMAL']: '[1,1,0]', @@ -338,7 +344,7 @@ class BasePlugin: 'X-Gizwits-Application-Id': 'c70a66ff039d41b4a220e198b0fcc8b3', } data = '{"raw": '+mode_str[mode]+'}' - url = 'https://euapi.gizwits.com/app/control/{}'.format(self.did) + url = f"https://euapi.gizwits.com/app/control/{self.did}" try: response = requests.post(url, headers=headers, data=data, timeout=3).json() except Exception as exc: @@ -350,10 +356,10 @@ class BasePlugin: Domoticz.Debug("Set Mode Response:" + str(response)) - if response != None: + if response is not 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])) + Domoticz.Status(f"New Heatzy Mode: {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) @@ -367,7 +373,7 @@ class BasePlugin: 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)) + Domoticz.Error(f"Cannot set Heatzy Mode: {error_message} ({error_code})\n{response}\nToken: {self.token}\nDeviceId: {self.did}") if error_code == 9004: #Invalid token self.token = "" @@ -379,36 +385,37 @@ class BasePlugin: return self.mode - def onOff(self, command): + def on_off(self, command): + """Toggle device on/off""" if Devices[1].sValue != command: if command == "On": - self.mode = self.setMode(HEATZY_MODE_VALUE['NORMAL']) + self.mode = self.set_mode(HEATZY_MODE_VALUE['NORMAL']) else: if not self.bug: - self.mode = self.setMode(HEATZY_MODE_VALUE['OFF']) + self.mode = self.set_mode(HEATZY_MODE_VALUE['OFF']) else: #Because of issue with the equipment (Off do not work...) - self.mode = self.setMode(HEATZY_MODE_VALUE['FROSTFREE']) + self.mode = self.set_mode(HEATZY_MODE_VALUE['FROSTFREE']) return self.mode -global _plugin _plugin = BasePlugin() -def onStart(): - global _plugin - _plugin.onStart() +def onStart(): #NOSONAR #pylint: disable=invalid-name + """OnStart""" + _plugin.on_start() -def onCommand(Unit, Command, Level, Hue): - global _plugin - _plugin.onCommand(Unit, Command, Level, Hue) +def onCommand(Unit, Command, Level, Hue): #NOSONAR #pylint: disable=invalid-name + """OnCommand""" + _plugin.on_command(Unit, Command, Level, Hue) -def onHeartbeat(): - global _plugin - _plugin.onHeartbeat() +def onHeartbeat(): #NOSONAR #pylint: disable=invalid-name + """onHeartbeat""" + _plugin.on_heartbeat() # Generic helper functions -def DumpConfigToLog(): +def dump_config_to_log(): + """Dump the config to the Domoticz Log""" for x in Parameters: if Parameters[x] != "": Domoticz.Debug( "'" + x + "':'" + str(Parameters[x]) + "'")