commit 7a1944966282f106d92cf94f10c1187967d343b6 Author: François JUMELLE Date: Mon May 24 10:33:49 2021 +0200 first commit diff --git a/plugin.py b/plugin.py new file mode 100755 index 0000000..6acdcf2 --- /dev/null +++ b/plugin.py @@ -0,0 +1,294 @@ +# Hygrostat python plugin for Domoticz +# +# Author: fjumelle +# +""" + + +

Hygrostat


+ Implementation of an hygrostat as a Domoticz Plugin.
+ Planning shall be configure like that: '[Monday]/.../[Sunday]' where each day is defined with 'Start-End' (Start and End format is HH:mm). + + + + + + + + + + + + + + + + + +""" +import Domoticz +import json +import urllib.parse as parse +import urllib.request as request +import base64 +import time +from datetime import datetime + +DEFAULT_POOLING = 30 #multiple of 15 sec +DEFAULT_DURATION = 30 +DEFAULT_RULE = "h_in > 70" + +class deviceparam: + def __init__(self, unit, nvalue, svalue): + self.unit = unit + self.nvalue = nvalue + self.svalue = svalue + + +class BasePlugin: + def __init__(self): + import math + + # Debug + self.debug = False + # Pooling + self.pooling_steps = math.ceil(DEFAULT_POOLING/15) + self.pooling = DEFAULT_POOLING // self.pooling_steps + self.pooling_current_step = 1 + + # Switch Id + self.switch_id = 0 + # Devive Ids + self.in_id = 0 + self.out_id = 0 + # Rule + self.rule = False + # Min duration (min) + self.min_duration = DEFAULT_DURATION + # Planning + self.planning = [("00:00", "23:59"), ("00:00", "23:59"), ("00:00", "23:59"), ("00:00", "23:59"), ("00:00", "23:59"), ("00:00", "23:59"), ("00:00", "23:59")] # From Monday to Sunday + # Current mode + self.mode = False #Off + # Last time rule is True or switch manually on + self.last_time = 0 + self.delay_in_progress = False + # N last values of huminity in + self.histo_hum = [] + return + + def onStart(self): + # 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 + Domoticz.Heartbeat(self.pooling) + + # Switch Id + try: + idx = int(Parameters["Mode1"]) + except: + raise Exception("Incorrect Switch idx") + self.switch_id = idx + + # Indoor/Outdoor Id + try: + in_idx = int(Parameters["Mode2"].split("/")[0]) + out_idx = int(Parameters["Mode2"].split("/")[1]) + except: + raise Exception("Incorrect Indoor/Outdoor Idx") + self.in_id = in_idx + self.out_id = out_idx + + # Rule + try: + rule = Parameters["Mode3"] + except: + rule = DEFAULT_RULE + self.rule = rule + + # Min duration + try: + duration = int(Parameters["Mode4"]) + except: + duration = DEFAULT_DURATION + self.min_duration = duration + + # Read planning + self.planning = [] + for day in Parameters["Mode5"].split("/"): + start_end = day.split("-", 1) + self.planning.append((start_end[0], start_end[1])) + if len(self.planning) != 7: + raise Exception("Incorrect planning...") + + + def onHeartbeat(self): + if self.pooling_current_step == self.pooling_steps: + #Now + now = time.time() + weekday = datetime.today().weekday() + + # Get device values + n_in, t_in, h_in, dp_in = get_temp_devide_info(self.in_id) + n_out, t_out, h_out, dp_out = get_temp_devide_info(self.out_id) + # Get switch values + n_sw, s_sw = get_switch_device_info(self.switch_id) + + #Keep last values (3 minutes) of indoor humidity + self.histo_hum.append(float(h_in)) + while len(self.histo_hum) > int(3 * 60 / DEFAULT_POOLING): + self.histo_hum.pop(0) + 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: + #Some one manually swtch on the device + #We keep the time + self.last_time = now + self.delay_in_progress = True + 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) + Domoticz.Debug("Start delay={}".format(now)) + + # If % huminidy > MAX ==> ON + if check_rule(self.rule, t_in, h_in, dp_in, t_out, h_out, dp_out, self.histo_hum): + self.mode = True #On + #We also keep the time, but only if in the authorized time range + if is_between(time.strftime("%H:%M", time.localtime(now)), self.planning[weekday]): + self.last_time = now + Domoticz.Debug("Condition satisfied ==> ON") + Domoticz.Debug("Start delay={}".format(now)) + else: + self.mode = False #Off + Domoticz.Debug("Condition satisfied but out of authorized time range ==> OFF") + else: + self.mode = False #Off + Domoticz.Debug("Condition not satisfied ==> OFF") + + if self.mode == True: + #Switch 'On' immediately if not the time range + if is_between(time.strftime("%H:%M", time.localtime(now)), self.planning[weekday]): + 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) + switch_on_off(self.switch_id, self.mode) + elif self.mode == False and now - self.last_time > self.min_duration*60: + #Switch 'Off' only after the delay + 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) + switch_on_off(self.switch_id, self.mode) + self.delay_in_progress = False + else: + Domoticz.Log("Delay not expired.") + Domoticz.Debug("Last Time={}".format(self.last_time)) + Domoticz.Debug("Now={}".format(now)) + Domoticz.Debug("Delta (s)={}".format(now - self.last_time)) + Domoticz.Debug("Delay (s)={}".format(self.min_duration*60)) + self.delay_in_progress = True + + self.pooling_current_step = 1 + else: + self.pooling_current_step = self.pooling_current_step + 1 + +global _plugin +_plugin = BasePlugin() + +def onStart(): + global _plugin + _plugin.onStart() + +def onHeartbeat(): + global _plugin + _plugin.onHeartbeat() + +def get_temp_devide_info(idx): + res = DomoticzAPI("type=devices&rid={0}".format(idx)) + name = res['result'][0]['Name'] + temp = res['result'][0]['Temp'] + try: + hum = res['result'][0]['Humidity'] + except: + hum = 0 + try: + dewpoint = res['result'][0]['DewPoint'] + except: + dewpoint = -100 + Domoticz.Debug("Device #{}: {} / T={}°C / H={}% / DP={}°C".format(idx, name, temp, hum, dewpoint)) + return name, float(temp), float(hum), float(dewpoint) + +def get_switch_device_info(idx): + res = DomoticzAPI("type=devices&rid={0}".format(idx)) + name = res['result'][0]['Name'] + status = False if res['result'][0]['Status'] == "Off" else True + Domoticz.Debug("Device #{}: {} / Status={}".format(idx, name, status)) + return name, status + +def switch_on_off(idx, mode=0): + # mode = False ==> OFF + # mode = True ==> ON + cmd = "Off" if mode == False else "On" + res = DomoticzAPI("type=command¶m=switchlight&idx={0}&switchcmd={1}".format(idx, cmd)) + Domoticz.Status("Switch #{} is now '{}'.".format(idx, cmd)) + return + +def check_rule(exp, t_in, h_in, dp_in, t_out, h_out, dp_out, histo_hum): + h_in_delta = float(h_in) - min(histo_hum) + res = eval(exp) + Domoticz.Debug("Check rule: {} ==> {}".format(exp, res)) + return res + +def is_between(time, time_range): + if time_range[1] < time_range[0]: + return time >= time_range[0] or time < time_range[1] + return time_range[0] <= time < 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): + Domoticz.Status("Indoor: {} / T={}°C / H={}% ({}) / DP={:.1f}°C".format(n_in, t_in, h_in, histo_h_in, dp_in)) + Domoticz.Status("Outdoor: {} / T={}°C / H={}% / DP={:.1f}°C".format(n_out, t_out, h_out, dp_out)) + +# 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 + +def DomoticzAPI(APICall): + resultJson = None + url = "http://{}:{}/json.htm?{}".format(Parameters["Address"], Parameters["Port"], parse.quote(APICall, safe="&=")) + Domoticz.Debug("Calling domoticz API: {}".format(url)) + try: + req = request.Request(url) + if Parameters["Username"] != "": + Domoticz.Debug("Add authentification for user {}".format(Parameters["Username"])) + credentials = ('%s:%s' % (Parameters["Username"], Parameters["Password"])) + encoded_credentials = base64.b64encode(credentials.encode('ascii')) + req.add_header('Authorization', 'Basic %s' % encoded_credentials.decode("ascii")) + + response = request.urlopen(req) + if response.status == 200: + resultJson = json.loads(response.read().decode('utf-8')) + if resultJson["status"] != "OK": + Domoticz.Error("Domoticz API returned an error: status = {}".format(resultJson["status"])) + resultJson = None + else: + Domoticz.Error("Domoticz API: http error = {}".format(response.status)) + except Exception as err: + Domoticz.Error("Error calling '{}'".format(url)) + Domoticz.Error(str(err)) + return resultJson