Refactor the source code to remove warnings

This commit is contained in:
Francois JUMELLE
2022-10-18 14:23:43 +02:00
parent f773665ead
commit 14be164024

165
plugin.py
View File

@@ -2,6 +2,8 @@
# #
# Author: fjumelle # Author: fjumelle
# #
#pylint: disable=line-too-long, invalid-name, undefined-variable, global-at-module-level
""" """
<plugin key="Hygrostat" name="Hygrostat" author="fjumelle" version="1.0.0" wikilink="" externallink=""> <plugin key="Hygrostat" name="Hygrostat" author="fjumelle" version="1.0.0" wikilink="" externallink="">
<description> <description>
@@ -28,29 +30,27 @@
</params> </params>
</plugin> </plugin>
""" """
import Domoticz
import json import json
import urllib.parse as parse import urllib.parse as parse
import urllib.request as request import urllib.request as request
import base64 import base64
import time import time
import math
from datetime import datetime, timedelta from datetime import datetime, timedelta
import Domoticz #pylint: disable=import-error
global Parameters
global Devices
DEFAULT_POOLING = 30 #multiple of 15 sec DEFAULT_POOLING = 30 #multiple of 15 sec
DEFAULT_DURATION = 30 DEFAULT_DURATION = 30
DEFAULT_RULE = "h_in > 70" DEFAULT_RULE = "h_in > 70"
class deviceparam: class BasePlugin(object):
def __init__(self, unit, nvalue, svalue): """Base class"""
self.unit = unit
self.nvalue = nvalue
self.svalue = svalue
class BasePlugin:
def __init__(self): def __init__(self):
import math """Creator"""
# Debug # Debug
self.debug = False self.debug = False
# Pooling # Pooling
@@ -76,15 +76,15 @@ class BasePlugin:
self.delay_in_progress = False self.delay_in_progress = False
# N last values of huminity in # N last values of huminity in
self.histo_hum = [] self.histo_hum = []
return
def onStart(self): def onStart(self): #NOSONAR
"""Plugin startup"""
# setup the appropriate logging level # setup the appropriate logging level
debuglevel = int(Parameters["Mode6"]) debuglevel = int(Parameters["Mode6"])
if debuglevel != 0: if debuglevel != 0:
self.debug = True self.debug = True
Domoticz.Debugging(debuglevel) Domoticz.Debugging(debuglevel)
DumpConfigToLog() dump_config_to_log()
else: else:
self.debug = False self.debug = False
Domoticz.Debugging(0) Domoticz.Debugging(0)
@@ -95,30 +95,30 @@ class BasePlugin:
# Switch Id # Switch Id
try: try:
idx = int(Parameters["Mode1"]) idx = int(Parameters["Mode1"])
except: except Exception as exc:
raise Exception("Incorrect Switch idx") raise ValueError("Incorrect Switch idx") from exc
self.switch_id = idx self.switch_id = idx
# Indoor/Outdoor Id # Indoor/Outdoor Id
try: try:
in_idx = int(Parameters["Mode2"].split("/")[0]) in_idx = int(Parameters["Mode2"].split("/")[0])
out_idx = int(Parameters["Mode2"].split("/")[1]) out_idx = int(Parameters["Mode2"].split("/")[1])
except: except Exception as exc:
raise Exception("Incorrect Indoor/Outdoor Idx") raise ValueError("Incorrect Indoor/Outdoor Idx") from exc
self.in_id = in_idx self.in_id = in_idx
self.out_id = out_idx self.out_id = out_idx
# Rule # Rule
try: try:
rule = Parameters["Mode3"] rule = Parameters["Mode3"]
except: except Exception: #pylint: disable=broad-except
rule = DEFAULT_RULE rule = DEFAULT_RULE
self.rule = rule self.rule = rule
# Min duration # Min duration
try: try:
duration = int(Parameters["Mode4"]) duration = int(Parameters["Mode4"])
except: except Exception: #pylint: disable=broad-except
duration = DEFAULT_DURATION duration = DEFAULT_DURATION
self.min_duration = duration self.min_duration = duration
@@ -128,34 +128,35 @@ class BasePlugin:
start_end = day.split("-", 1) start_end = day.split("-", 1)
self.planning.append((start_end[0], start_end[1])) self.planning.append((start_end[0], start_end[1]))
if len(self.planning) != 7: if len(self.planning) != 7:
raise Exception("Incorrect planning...") raise ValueError("Incorrect planning...")
def onHeartbeat(self): def onHeartbeat(self): #NOSONAR
"""Plugin heartbeat"""
if self.pooling_current_step == self.pooling_steps: if self.pooling_current_step == self.pooling_steps:
#Now #Now
now = time.time() now = time.time()
weekday = datetime.today().weekday() weekday = datetime.today().weekday()
# Get device values # Get device values
n_in, t_in, h_in, dp_in, lu_in = get_temp_devide_info(self.in_id) n_in, t_in, h_in, dp_in, lu_in = get_temp_devide_info(self.in_id)
n_out, t_out, h_out, dp_out, lu_out = get_temp_devide_info(self.out_id) n_out, t_out, h_out, dp_out, lu_out = get_temp_devide_info(self.out_id)
# Get switch values # Get switch values
n_sw, s_sw = get_switch_device_info(self.switch_id) _, s_sw = get_switch_device_info(self.switch_id)
#Keep last values (3 minutes) of indoor humidity #Keep last values (3 minutes) of indoor humidity
self.histo_hum.append(float(h_in)) self.histo_hum.append(float(h_in))
while len(self.histo_hum) > int(3 * 60 / DEFAULT_POOLING): while len(self.histo_hum) > int(3 * 60 / DEFAULT_POOLING):
self.histo_hum.pop(0) self.histo_hum.pop(0)
Domoticz.Debug("Last 3 minutes of indoor humidity: " + str(self.histo_hum)) Domoticz.Debug("Last 3 minutes of indoor humidity: " + str(self.histo_hum))
if self.delay_in_progress == False and self.mode == False and s_sw == True: if not self.delay_in_progress and not self.mode and s_sw:
#Someone manually switched on the device #Someone manually switched on the device
#We keep the time #We keep the time
self.last_time = now self.last_time = now
self.delay_in_progress = True self.delay_in_progress = True
Domoticz.Status("Switch on manually") Domoticz.Status("Switch on manually")
print_status(self.in_id, n_in, t_in, h_in, dp_in, self.histo_hum, self.out_id, n_out, t_out, h_out, dp_out) print_status(self.in_id, n_in, t_in, h_in, dp_in, self.histo_hum, self.out_id, n_out, t_out, h_out, dp_out)
Domoticz.Debug("Start delay={}".format(now)) Domoticz.Debug(f"Start delay={now}")
# If % huminidy > MAX ==> ON # If % huminidy > MAX ==> ON
if check_rule(self.rule, t_in, h_in, dp_in, lu_in, t_out, h_out, dp_out, lu_out, self.histo_hum): if check_rule(self.rule, t_in, h_in, dp_in, lu_in, t_out, h_out, dp_out, lu_out, self.histo_hum):
@@ -164,7 +165,7 @@ class BasePlugin:
if is_between(time.strftime("%H:%M", time.localtime(now)), self.planning[weekday]): if is_between(time.strftime("%H:%M", time.localtime(now)), self.planning[weekday]):
self.last_time = now self.last_time = now
Domoticz.Debug("Condition satisfied ==> ON") Domoticz.Debug("Condition satisfied ==> ON")
Domoticz.Debug("Start delay={}".format(now)) Domoticz.Debug(f"Start delay={now}")
else: else:
self.mode = False #Off self.mode = False #Off
Domoticz.Debug("Condition satisfied but out of authorized time range ==> OFF") Domoticz.Debug("Condition satisfied but out of authorized time range ==> OFF")
@@ -172,13 +173,13 @@ class BasePlugin:
self.mode = False #Off self.mode = False #Off
Domoticz.Debug("Condition not satisfied ==> OFF") Domoticz.Debug("Condition not satisfied ==> OFF")
if self.mode == True: if self.mode:
#Switch 'On' immediately if not the time range #Switch 'On' immediately if not the time range
if is_between(time.strftime("%H:%M", time.localtime(now)), self.planning[weekday]): if is_between(time.strftime("%H:%M", time.localtime(now)), self.planning[weekday]):
if self.mode != s_sw: if self.mode != s_sw:
print_status(self.in_id, n_in, t_in, h_in, dp_in, self.histo_hum, self.out_id, n_out, t_out, h_out, dp_out) print_status(self.in_id, n_in, t_in, h_in, dp_in, self.histo_hum, self.out_id, n_out, t_out, h_out, dp_out)
switch_on_off(self.switch_id, self.mode) switch_on_off(self.switch_id, self.mode)
elif self.mode == False and now - self.last_time > self.min_duration*60: elif not self.mode and now - self.last_time > self.min_duration*60:
#Switch 'Off' only after the delay #Switch 'Off' only after the delay
if self.mode != s_sw: if self.mode != s_sw:
print_status(self.in_id, n_in, t_in, h_in, dp_in, self.histo_hum, self.out_id, n_out, t_out, h_out, dp_out) print_status(self.in_id, n_in, t_in, h_in, dp_in, self.histo_hum, self.out_id, n_out, t_out, h_out, dp_out)
@@ -186,78 +187,84 @@ class BasePlugin:
self.delay_in_progress = False self.delay_in_progress = False
else: else:
Domoticz.Log("Delay not expired.") Domoticz.Log("Delay not expired.")
Domoticz.Debug("Last Time={}".format(self.last_time)) Domoticz.Debug(f"Last Time={self.last_time}")
Domoticz.Debug("Now={}".format(now)) Domoticz.Debug(f"Now={now}")
Domoticz.Debug("Delta (s)={}".format(now - self.last_time)) Domoticz.Debug(f"Delta (s)={now - self.last_time}")
Domoticz.Debug("Delay (s)={}".format(self.min_duration*60)) Domoticz.Debug(f"Delay (s)={self.min_duration*60}")
self.delay_in_progress = True self.delay_in_progress = True
self.pooling_current_step = 1 self.pooling_current_step = 1
else: else:
self.pooling_current_step = self.pooling_current_step + 1 self.pooling_current_step = self.pooling_current_step + 1
global _plugin global _plugin
_plugin = BasePlugin() _plugin = BasePlugin()
def onStart(): def onStart(): #NOSONAR
global _plugin """Plugin start"""
_plugin.onStart() _plugin.onStart()
def onHeartbeat(): def onHeartbeat(): #NOSONAR
global _plugin """Plugin heartbeat"""
_plugin.onHeartbeat() _plugin.onHeartbeat()
def get_temp_devide_info(idx): def get_temp_devide_info(idx):
res = DomoticzAPI("type=devices&rid={0}".format(idx)) """Get data from temp/hum devide idx"""
res = domoticz_api(f"type=devices&rid={idx}")
name = res['result'][0]['Name'] name = res['result'][0]['Name']
temp = res['result'][0]['Temp'] temp = res['result'][0]['Temp']
lastUpdate = res['result'][0]['LastUpdate'] last_update = res['result'][0]['LastUpdate']
try: try:
hum = res['result'][0]['Humidity'] hum = res['result'][0]['Humidity']
except: except Exception: #pylint: disable=broad-except
hum = 0 hum = 0
try: try:
dewpoint = res['result'][0]['DewPoint'] dewpoint = res['result'][0]['DewPoint']
except: except Exception: #pylint: disable=broad-except
dewpoint = -100 dewpoint = -100
Domoticz.Debug("Device #{}: {} / T={}°C / H={}% / DP={}°C ({})".format(idx, name, temp, hum, dewpoint, lastUpdate)) Domoticz.Debug(f"Device #{idx}: {name} / T={temp}°C / H={hum}% / DP={dewpoint}°C ({last_update})")
return name, float(temp), float(hum), float(dewpoint), str(lastUpdate) return name, float(temp), float(hum), float(dewpoint), str(last_update)
def get_switch_device_info(idx): def get_switch_device_info(idx):
res = DomoticzAPI("type=devices&rid={0}".format(idx)) """Get data from switch devide idx"""
res = domoticz_api(f"type=devices&rid={idx}")
name = res['result'][0]['Name'] name = res['result'][0]['Name']
status = False if res['result'][0]['Status'] == "Off" else True status = False if res['result'][0]['Status'] == "Off" else True
Domoticz.Debug("Device #{}: {} / Status={}".format(idx, name, status)) Domoticz.Debug(f"Device #{idx}: {name} / Status={status}")
return name, status return name, status
def switch_on_off(idx, mode=0): def switch_on_off(idx, mode=0):
"""Toggle switch device idx"""
# mode = False ==> OFF # mode = False ==> OFF
# mode = True ==> ON # mode = True ==> ON
cmd = "Off" if mode == False else "On" cmd = "Off" if not mode else "On"
res = DomoticzAPI("type=command&param=switchlight&idx={0}&switchcmd={1}".format(idx, cmd)) domoticz_api(f"type=command&param=switchlight&idx={idx}&switchcmd={cmd}")
Domoticz.Status("Switch #{} is now '{}'.".format(idx, cmd)) Domoticz.Status(f"Switch #{idx} is now '{cmd}'.")
return
def check_rule(exp, t_in, h_in, dp_in, lu_in, t_out, h_out, dp_out, lu_out, histo_hum): def check_rule(exp, t_in, h_in, dp_in, lu_in, t_out, h_out, dp_out, lu_out, histo_hum): #pylint: disable=unused-argument #NOSONAR
"""Check the rule"""
if lu_in<(datetime.now()-timedelta(minutes=DEFAULT_DURATION)).strftime("%Y-%m-%d %H:%M:%S"): if lu_in<(datetime.now()-timedelta(minutes=DEFAULT_DURATION)).strftime("%Y-%m-%d %H:%M:%S"):
Domoticz.Status("Device In seems obsolete ({})".format(lu_in)) Domoticz.Status(f"Device In seems obsolete ({lu_in})")
return False return False
h_in_delta = float(h_in) - min(histo_hum) h_in_delta = float(h_in) - min(histo_hum) #pylint: disable=unused-variable #NOSONAR
res = eval(exp) res = eval(exp) #pylint: disable=eval-used
Domoticz.Debug("Check rule: {} ==> {}".format(exp, res)) Domoticz.Debug(f"Check rule: {exp} ==> {res}")
return res return res
def is_between(time, time_range): def is_between(time_current, time_range):
"""Check if time_current is in the time range"""
if time_range[1] < time_range[0]: if time_range[1] < time_range[0]:
return time >= time_range[0] or time < time_range[1] return time_current >= time_range[0] or time_current < time_range[1]
return time_range[0] <= time < time_range[1] return time_range[0] <= time_current < time_range[1]
def print_status(idx_in, n_in, t_in, h_in, dp_in, histo_h_in, idx_out, n_out, t_out, h_out, dp_out): def print_status(idx_in, n_in, t_in, h_in, dp_in, histo_h_in, idx_out, n_out, t_out, h_out, dp_out):
Domoticz.Status("Indoor: {} / T={}°C / H={}% ({}) / DP={:.1f}°C".format(n_in, t_in, h_in, histo_h_in, dp_in)) """Print indoor/outdoor status"""
Domoticz.Status("Outdoor: {} / T={}°C / H={}% / DP={:.1f}°C".format(n_out, t_out, h_out, dp_out)) Domoticz.Status(f"Indoor (#{idx_in}): {n_in} / T={t_in}°C / H={h_in}% ({histo_h_in}) / DP={dp_in:.1f}°C")
Domoticz.Status(f"Outdoor (#{idx_out}): {n_out} / T={t_out}°C / H={h_out}% / DP={dp_out:.1f}°C")
# Generic helper functions # Generic helper functions
def DumpConfigToLog(): def dump_config_to_log():
"""Dump the plugin config to the domoticz log"""
for x in Parameters: for x in Parameters:
if Parameters[x] != "": if Parameters[x] != "":
Domoticz.Debug( "'" + x + "':'" + str(Parameters[x]) + "'") Domoticz.Debug( "'" + x + "':'" + str(Parameters[x]) + "'")
@@ -269,29 +276,29 @@ def DumpConfigToLog():
Domoticz.Debug("Device nValue: " + str(Devices[x].nValue)) Domoticz.Debug("Device nValue: " + str(Devices[x].nValue))
Domoticz.Debug("Device sValue: '" + Devices[x].sValue + "'") Domoticz.Debug("Device sValue: '" + Devices[x].sValue + "'")
Domoticz.Debug("Device LastLevel: " + str(Devices[x].LastLevel)) Domoticz.Debug("Device LastLevel: " + str(Devices[x].LastLevel))
return
def DomoticzAPI(APICall): def domoticz_api(params):
resultJson = None """Call the Domoticz API"""
url = "http://{}:{}/json.htm?{}".format(Parameters["Address"], Parameters["Port"], parse.quote(APICall, safe="&=")) result_json = None
Domoticz.Debug("Calling domoticz API: {}".format(url)) url = f"http://{Parameters['Address']}:{Parameters['Port']}/json.htm?{parse.quote(params, safe='&=')}"
Domoticz.Debug(f"Calling domoticz API: {url}")
try: try:
req = request.Request(url) req = request.Request(url)
if Parameters["Username"] != "": if Parameters["Username"] != "":
Domoticz.Debug("Add authentification for user {}".format(Parameters["Username"])) Domoticz.Debug(f"Add authentification for user {Parameters['Username']}")
credentials = ('%s:%s' % (Parameters["Username"], Parameters["Password"])) credentials = f"{Parameters['Username']}:{Parameters['Password']}"
encoded_credentials = base64.b64encode(credentials.encode('ascii')) encoded_credentials = base64.b64encode(credentials.encode('ascii'))
req.add_header('Authorization', 'Basic %s' % encoded_credentials.decode("ascii")) req.add_header("Authorization", f"Basic {encoded_credentials.decode('ascii')}")
response = request.urlopen(req) response = request.urlopen(req)
if response.status == 200: if response.status == 200:
resultJson = json.loads(response.read().decode('utf-8')) result_json = json.loads(response.read().decode('utf-8'))
if resultJson["status"] != "OK": if result_json["status"] != "OK":
Domoticz.Error("Domoticz API returned an error: status = {}".format(resultJson["status"])) Domoticz.Error(f"Domoticz API returned an error: status = {result_json['status']}")
resultJson = None result_json = None
else: else:
Domoticz.Error("Domoticz API: http error = {}".format(response.status)) Domoticz.Error(f"Domoticz API: http error = {response.status}")
except Exception as err: except Exception as exc: #pylint: disable=broad-except
Domoticz.Error("Error calling '{}'".format(url)) Domoticz.Error(f"Error calling '{url}'")
Domoticz.Error(str(err)) Domoticz.Error(str(exc))
return resultJson return result_json