316 lines
12 KiB
Python
Executable File
316 lines
12 KiB
Python
Executable File
# Hygrostat python plugin for Domoticz
|
|
#
|
|
# Author: fjumelle
|
|
#
|
|
|
|
#pylint: disable=line-too-long, invalid-name, undefined-variable, global-at-module-level
|
|
"""
|
|
<plugin key="Hygrostat" name="Hygrostat" author="fjumelle" version="2.0.0" wikilink="" externallink="">
|
|
<description>
|
|
<h2>Hygrostat</h2><br/>
|
|
Implementation of an hygrostat as a Domoticz Plugin.
|
|
</description>
|
|
<params>
|
|
<param field="Address" label="Domoticz IP Address" width="200px" required="true" default="localhost"/>
|
|
<param field="Port" label="Port" width="40px" required="true" default="8080"/>
|
|
<param field="Username" label="Domoticz Username" width="200px" required="false" default=""/>
|
|
<param field="Password" label="Domoticz Password" width="200px" required="false" default="" password="true"/>
|
|
<param field="Mode1" label="Switch idx" width="200px"/>
|
|
<param field="Mode2" label="In/Out idx" width="200px"/>
|
|
<param field="Mode3" label="Rule" width="600px"/>
|
|
<param field="Mode4" label="Min duration (min)" width="200px"/>
|
|
<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 json
|
|
import urllib.parse as parse
|
|
import urllib.request as request
|
|
import base64
|
|
import time
|
|
import math
|
|
from datetime import datetime, timedelta
|
|
|
|
import Domoticz #pylint: disable=import-error
|
|
|
|
global Parameters
|
|
global Devices
|
|
|
|
DEFAULT_POOLING = 30 #multiple of 15 sec
|
|
DEFAULT_DURATION = 30
|
|
DEFAULT_RULE = "h_in > 70"
|
|
|
|
DEVICE_UNIT = 1 #Id of the device crerated by the module
|
|
|
|
class BasePlugin(object):
|
|
"""Base class"""
|
|
def __init__(self):
|
|
"""Creator"""
|
|
# 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
|
|
# 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 = []
|
|
|
|
def onStart(self): #NOSONAR
|
|
"""Plugin startup"""
|
|
# setup the appropriate logging level
|
|
debuglevel = int(Parameters["Mode6"])
|
|
if debuglevel != 0:
|
|
self.debug = True
|
|
Domoticz.Debugging(debuglevel)
|
|
dump_config_to_log()
|
|
else:
|
|
self.debug = False
|
|
Domoticz.Debugging(0)
|
|
|
|
# Polling interval = X sec
|
|
Domoticz.Heartbeat(self.pooling)
|
|
|
|
# Switch Id
|
|
try:
|
|
idx = int(Parameters["Mode1"])
|
|
except Exception as exc:
|
|
raise ValueError("Incorrect Switch idx") from exc
|
|
self.switch_id = idx
|
|
|
|
# Indoor/Outdoor Id
|
|
try:
|
|
in_idx = int(Parameters["Mode2"].split("/")[0])
|
|
out_idx = int(Parameters["Mode2"].split("/")[1])
|
|
except Exception as exc:
|
|
raise ValueError("Incorrect Indoor/Outdoor Idx") from exc
|
|
self.in_id = in_idx
|
|
self.out_id = out_idx
|
|
|
|
# Rule
|
|
try:
|
|
rule = Parameters["Mode3"]
|
|
except Exception: #pylint: disable=broad-except
|
|
rule = DEFAULT_RULE
|
|
self.rule = rule
|
|
|
|
# Min duration
|
|
try:
|
|
duration = int(Parameters["Mode4"])
|
|
except Exception: #pylint: disable=broad-except
|
|
duration = DEFAULT_DURATION
|
|
self.min_duration = duration
|
|
|
|
#Create the device(s) if needed
|
|
if DEVICE_UNIT not in Devices:
|
|
#Create the Pause device
|
|
Domoticz.Device(Unit=DEVICE_UNIT,
|
|
Name="Pause",
|
|
TypeName="Switch",
|
|
Image=9,
|
|
Used=1
|
|
).Create()
|
|
|
|
|
|
def onHeartbeat(self): #NOSONAR
|
|
"""Plugin heartbeat"""
|
|
if self.pooling_current_step == self.pooling_steps:
|
|
#Now
|
|
now = time.time()
|
|
|
|
# Get device values
|
|
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)
|
|
# Get switch values
|
|
_, 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 not self.delay_in_progress and not self.mode and s_sw:
|
|
#Someone manually switched 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(f"Start delay={now}")
|
|
|
|
# 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):
|
|
self.mode = True #On
|
|
#We also keep the time, but only if in the authorized time range
|
|
if Devices[DEVICE_UNIT].nValue == 0:
|
|
self.last_time = now
|
|
Domoticz.Debug("Condition satisfied ==> ON")
|
|
Domoticz.Debug(f"Start delay={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:
|
|
#Switch 'On' immediately if not the time range
|
|
if Devices[DEVICE_UNIT].nValue == 0 and 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 not self.mode 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(f"Last Time={self.last_time}")
|
|
Domoticz.Debug(f"Now={now}")
|
|
Domoticz.Debug(f"Delta (s)={now - self.last_time}")
|
|
Domoticz.Debug(f"Delay (s)={self.min_duration*60}")
|
|
self.delay_in_progress = True
|
|
|
|
self.pooling_current_step = 1
|
|
else:
|
|
self.pooling_current_step = self.pooling_current_step + 1
|
|
|
|
def onCommand(self, Unit, Command, Level, Color): #pylint: disable=unused-argument #NOSONAR
|
|
"""Plugin command"""
|
|
Domoticz.Debug(f"onCommand called for Unit {Unit}: Command '{Command}', Level: {Level}")
|
|
|
|
if Unit == DEVICE_UNIT: # pause switch
|
|
svalue = str(Command)
|
|
if str(Command) == "On":
|
|
nvalue = 1
|
|
else:
|
|
nvalue = 0
|
|
|
|
Devices[Unit].Update(nValue=nvalue, sValue=svalue)
|
|
|
|
|
|
global _plugin
|
|
_plugin = BasePlugin()
|
|
|
|
def onStart(): #NOSONAR
|
|
"""Plugin start"""
|
|
_plugin.onStart()
|
|
|
|
def onHeartbeat(): #NOSONAR
|
|
"""Plugin heartbeat"""
|
|
_plugin.onHeartbeat()
|
|
|
|
def onCommand(Unit, Command, Level, Color): #NOSONAR
|
|
"""Plugin command"""
|
|
_plugin.onCommand(Unit, Command, Level, Color)
|
|
|
|
def get_temp_devide_info(idx):
|
|
"""Get data from temp/hum devide idx"""
|
|
res = domoticz_api(f"type=devices&rid={idx}")
|
|
name = res['result'][0]['Name']
|
|
temp = res['result'][0]['Temp']
|
|
last_update = res['result'][0]['LastUpdate']
|
|
try:
|
|
hum = res['result'][0]['Humidity']
|
|
except Exception: #pylint: disable=broad-except
|
|
hum = 0
|
|
try:
|
|
dewpoint = res['result'][0]['DewPoint']
|
|
except Exception: #pylint: disable=broad-except
|
|
dewpoint = -100
|
|
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(last_update)
|
|
|
|
def get_switch_device_info(idx):
|
|
"""Get data from switch devide idx"""
|
|
res = domoticz_api(f"type=devices&rid={idx}")
|
|
name = res['result'][0]['Name']
|
|
status = False if res['result'][0]['Status'] == "Off" else True
|
|
Domoticz.Debug(f"Device #{idx}: {name} / Status={status}")
|
|
return name, status
|
|
|
|
def switch_on_off(idx, mode=0):
|
|
"""Toggle switch device idx"""
|
|
# mode = False ==> OFF
|
|
# mode = True ==> ON
|
|
cmd = "Off" if not mode else "On"
|
|
domoticz_api(f"type=command¶m=switchlight&idx={idx}&switchcmd={cmd}")
|
|
Domoticz.Status(f"Switch #{idx} is now '{cmd}'.")
|
|
|
|
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"):
|
|
Domoticz.Status(f"Device In seems obsolete ({lu_in})")
|
|
return False
|
|
h_in_delta = float(h_in) - min(histo_hum) #pylint: disable=unused-variable #NOSONAR
|
|
res = eval(exp) #pylint: disable=eval-used
|
|
Domoticz.Debug(f"Check rule: {exp} ==> {res}")
|
|
return res
|
|
|
|
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):
|
|
"""Print indoor/outdoor status"""
|
|
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
|
|
def dump_config_to_log():
|
|
"""Dump the plugin config to the domoticz log"""
|
|
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))
|
|
|
|
def domoticz_api(params):
|
|
"""Call the Domoticz API"""
|
|
result_json = None
|
|
url = f"http://{Parameters['Address']}:{Parameters['Port']}/json.htm?{parse.quote(params, safe='&=')}"
|
|
Domoticz.Debug(f"Calling domoticz API: {url}")
|
|
try:
|
|
req = request.Request(url)
|
|
if Parameters["Username"] != "":
|
|
Domoticz.Debug(f"Add authentification for user {Parameters['Username']}")
|
|
credentials = f"{Parameters['Username']}:{Parameters['Password']}"
|
|
encoded_credentials = base64.b64encode(credentials.encode('ascii'))
|
|
req.add_header("Authorization", f"Basic {encoded_credentials.decode('ascii')}")
|
|
|
|
response = request.urlopen(req)
|
|
if response.status == 200:
|
|
result_json = json.loads(response.read().decode('utf-8'))
|
|
if result_json["status"] != "OK":
|
|
Domoticz.Error(f"Domoticz API returned an error: status = {result_json['status']}")
|
|
result_json = None
|
|
else:
|
|
Domoticz.Error(f"Domoticz API: http error = {response.status}")
|
|
except Exception as exc: #pylint: disable=broad-except
|
|
Domoticz.Error(f"Error calling '{url}'")
|
|
Domoticz.Error(str(exc))
|
|
return result_json
|