247 lines
14 KiB
Python
247 lines
14 KiB
Python
import appdaemon.plugins.hass.hassapi as hass
|
|
import ad_toolbox.smartcondition as SmartCondition
|
|
from ad_toolbox.smartobject import SmartObject
|
|
import time
|
|
|
|
class SmartShutter(SmartObject):
|
|
|
|
#@SmartCondition.catch_smartcondition_exception(lambda self, message: self.log_error(message,stop_app = True))
|
|
def on_initialize_smart_object(self):
|
|
#super().initialize()
|
|
|
|
self.slow_transition = False
|
|
self.slow_transition_start_time = 0
|
|
self.slow_transition_cb = None
|
|
self.target_position = None
|
|
self.target_position_reach_time = None
|
|
self.ready = False
|
|
self.open_conditions = None
|
|
self.close_conditions = None
|
|
|
|
if "auto_switch" in self.args:
|
|
self.auto_switch = self.args['auto_switch']
|
|
self.listen_state(self.on_auto_switch_on,self.auto_switch, old = 'off', new = 'on')
|
|
else:
|
|
self.auto_switch = None
|
|
|
|
if "open_position" in self.args:
|
|
try: self.open_position = int(self.args["open_position"])
|
|
except:
|
|
self.open_position_sensor = self.args["open_position"]
|
|
self.open_position = int(float(self.get_state(self.open_position_sensor)))
|
|
self.log(f"Registering sensor {self.open_position_sensor} for open position")
|
|
self.listen_state(self.on_open_position_change,self.open_position_sensor)
|
|
else:
|
|
self.open_position = 100
|
|
|
|
if "close_position" in self.args:
|
|
try: self.close_position = int(self.args["close_position"])
|
|
except:
|
|
self.close_position_sensor = self.args["close_position"]
|
|
self.close_position = int(float(self.get_state(self.close_position_sensor)))
|
|
self.log(f"Registering sensor {self.close_position_sensor} for close position")
|
|
self.listen_state(self.on_close_position_change,self.close_position_sensor)
|
|
else:
|
|
self.close_position = 0
|
|
|
|
#it need to be called before on_update
|
|
if "sleeping_switch" in self.args:
|
|
self.listen_state(self.on_stop_sleeping,self.args['sleeping_switch'], old = 'on', new = 'off')
|
|
|
|
extra_entities_to_listen = list()
|
|
|
|
if "emergency_close_conditions" in self.args:
|
|
self.emergency_close_conditions = SmartCondition.Evaluator(self,self.args["emergency_close_conditions"], condition_name = "emergency_close_conditions", templates_library = self.templates_library, constants = self.constants)
|
|
extra_entities_to_listen.extend(self.emergency_close_conditions.get_entities_to_listen_as_list())
|
|
else:
|
|
self.emergency_close_conditions = None
|
|
|
|
if 'close_conditions' in self.args:
|
|
self.close_conditions = SmartCondition.Evaluator(self,self.args["close_conditions"], condition_name = "close_conditions", templates_library = self.templates_library,on_update_cb = self.on_update,extra_entities_to_listen = extra_entities_to_listen, log_callback_trigger_reason = False, constants = self.constants)
|
|
if 'open_conditions' in self.args:
|
|
self.log("Both a close_conditions and an open_conditions has been defined, please use only one",level = 'ERROR')
|
|
elif 'open_conditions' in self.args:
|
|
self.open_conditions = SmartCondition.Evaluator(self,self.args["open_conditions"], condition_name = "open_conditions", templates_library = self.templates_library,on_update_cb = self.on_update,extra_entities_to_listen = extra_entities_to_listen, log_callback_trigger_reason = False, constants = self.constants)
|
|
else:
|
|
self.log("No close_conditions nor open_conditions defined",level = 'ERROR')
|
|
|
|
#for object that take a long time to change state, we need reevalute desired state once the state has changed
|
|
#in case the conditions change during the state change
|
|
#in case of shutter we can probably improve that and cancel the state change has it is updating
|
|
# it seems to messed up with the auto switch and I'm not sure it's relevant anyway
|
|
#self.listen_state(self.on_state_change,self.entity_id)
|
|
self.listen_state(self.on_position_change,self.entity_id,attribute = 'current_position')
|
|
|
|
self.ready = True
|
|
self.on_update()
|
|
|
|
def terminate(self):
|
|
self.emergency_close_conditions = None
|
|
self.close_conditions = None
|
|
self.open_conditions = None
|
|
super().terminate()
|
|
|
|
def on_state_change(self, entity, attribute, old, new, kwargs):
|
|
if self.ready == False:
|
|
self.log(f"on_state_change called before the app is ready",level = "WARNING")
|
|
return
|
|
#for object that take a long time to change state, we need reevalute desired state once the state has changed
|
|
#in case the conditions change during the state change
|
|
#in case of shutter we can probably improve that and cancel the state change has it is updating
|
|
if old != new and not self.slow_transition:
|
|
self.log("state changed from " + str(old) + " to " + str(new))
|
|
self.on_update()
|
|
|
|
def on_position_change(self, entity, attribute, old, new, kwargs):
|
|
if self.ready == False:
|
|
self.log(f"on_position_change called before the app is ready",level = "WARNING")
|
|
return
|
|
|
|
if self.target_position == None:
|
|
if self.target_position_reach_time != None and time.time() - self.target_position_reach_time < 60:
|
|
self.log(f"A position changed ({old} -> {new}) occured {time.time() - self.target_position_reach_time}s after a requested position change has been complete. This won't be considered as a manual command")
|
|
else:
|
|
self.log(f"Manual command has been received {old} -> {new}")
|
|
if self.close_conditions: smart_conditions = self.close_conditions
|
|
else: smart_conditions = self.open_conditions
|
|
|
|
if self.auto_switch and smart_conditions.evaluate(False) != SmartCondition.Result.Disabled:
|
|
self.set_state(self.auto_switch,state = "off")
|
|
else:
|
|
if self.target_position == new:
|
|
self.log(f"Movement completed old: {old}, new: {new}, target_position: {self.target_position}, current_position: {self.get_state(self.entity_id,attribute = 'current_position')}")
|
|
self.target_position = None
|
|
self.target_position_reach_time = time.time()
|
|
|
|
def on_open_position_change(self, entity, attribute, old, new, kwargs):
|
|
if self.ready == False:
|
|
self.log(f"on_open_position_change called before the app is ready",level = "WARNING")
|
|
return
|
|
|
|
if new != self.open_position:
|
|
self.open_position = int(float(new))
|
|
|
|
if abs(self.open_position - int(float(self.get_state(self.entity_id,attribute = "current_position")))) >= 5 or self.open_position == 100:
|
|
self.on_update()
|
|
|
|
def on_close_position_change(self, entity, attribute, old, new, kwargs):
|
|
if self.ready == False:
|
|
self.log(f"on_close_position_change called before the app is ready",level = "WARNING")
|
|
return
|
|
|
|
if new != self.close_position:
|
|
self.close_position = int(float(new))
|
|
|
|
if abs(self.close_position - int(float(self.get_state(self.entity_id,attribute = "current_position")))) >= 5 or self.close_position == 0:
|
|
self.on_update()
|
|
|
|
def on_stop_sleeping(self, entity, attribute, old, new, kwargs):
|
|
if self.get_state('sun.sun') != 'below_horizon':
|
|
self.slow_transition = True
|
|
self.slow_transition_start_time = self.datetime()
|
|
self.on_update()
|
|
|
|
def on_auto_switch_on(self, entity, attribute, old, new, kwargs):
|
|
self.on_update()
|
|
|
|
def is_auto_switch_on(self):
|
|
if self.auto_switch:
|
|
return self.get_state(self.auto_switch) == 'on'
|
|
else:
|
|
return True
|
|
|
|
def calculate_transition_position(self):
|
|
time_since_transition_start = self.datetime() - self.slow_transition_start_time
|
|
time_since_transition_start.total_seconds()
|
|
return min((time_since_transition_start.total_seconds() / 10) + 10,self.open_position)
|
|
|
|
def on_transition_update(self, kwargs):
|
|
self.slow_transition_cb = None
|
|
if self.slow_transition: self.on_update()
|
|
|
|
def stop_slow_transition(self):
|
|
self.slow_transition = False
|
|
if self.slow_transition_cb:
|
|
self.cancel_timer(self.slow_transition_cb)
|
|
self.slow_transition_cb = None
|
|
|
|
def on_retry_update(self, kwargs):
|
|
self.log("on_retry_update")
|
|
self.on_update()
|
|
|
|
def on_update(self):
|
|
def log_evaluation_result(smart_conditions):
|
|
smart_conditions.log_callback_trigger_reason()
|
|
smart_conditions.log_evaluation_result()
|
|
|
|
if self.emergency_close_conditions and self.emergency_close_conditions.evaluate(False) == SmartCondition.Result.Succeeded:
|
|
if self.get_state(self.entity_id) != 'closed':
|
|
log_evaluation_result(self.emergency_close_conditions)
|
|
self.log("Closing shutter (Emergency close)")
|
|
self.call_service("cover/close_cover", entity_id = self.entity_id)
|
|
if self.auto_switch: self.set_state(self.auto_switch,state = "on")
|
|
elif self.is_auto_switch_on():
|
|
if self.close_conditions: smart_conditions = self.close_conditions
|
|
else: smart_conditions = self.open_conditions
|
|
|
|
if smart_conditions: result = smart_conditions.evaluate(False)
|
|
else:
|
|
# it seems sometimes during HA reboot, the CB is called and both close_conditions and open_conditions are equal to None
|
|
# this should probably require more investigation
|
|
self.log_warning(f"Both close_conditions and open_conditions are None")
|
|
return
|
|
|
|
if result != SmartCondition.Result.Disabled:
|
|
if self.get_state(self.entity_id,attribute = "current_position") != None:
|
|
if self.close_conditions != None:
|
|
should_be_close = (result == SmartCondition.Result.Succeeded)
|
|
else:
|
|
should_be_close = (result == SmartCondition.Result.Failed)
|
|
current_position = int(self.get_state(self.entity_id,attribute = "current_position"))
|
|
current_state = self.get_state(self.entity_id)
|
|
|
|
#if current_position > 20 and should_be_close:
|
|
if should_be_close:
|
|
if self.close_position > 0:
|
|
if current_position != self.close_position:
|
|
log_evaluation_result(smart_conditions)
|
|
self.target_position = self.close_position
|
|
self.log("Closing shutter at " + str(self.close_position) + " (previous position was " + str(current_position) + ")")
|
|
self.call_service("cover/set_cover_position", entity_id = self.entity_id,position = self.close_position)
|
|
elif current_state != 'closed':
|
|
log_evaluation_result(smart_conditions)
|
|
self.target_position = self.close_position
|
|
self.log(f"Closing shutter, state was {current_state}")
|
|
self.call_service("cover/close_cover", entity_id = self.entity_id)
|
|
|
|
self.stop_slow_transition()
|
|
elif abs(current_position - self.open_position) > 5 and not should_be_close:
|
|
self.target_position = self.open_position
|
|
if self.slow_transition:
|
|
if self.slow_transition_cb == None:
|
|
wanted_position = self.calculate_transition_position()
|
|
if wanted_position > 60:
|
|
self.call_service("cover/open_cover", entity_id = self.entity_id)
|
|
self.stop_slow_transition()
|
|
else:
|
|
self.call_service("cover/set_cover_position", entity_id = self.entity_id,position = wanted_position)
|
|
self.slow_transition_cb = self.run_in(self.on_transition_update, 120)
|
|
else:
|
|
self.log(f"Update ignored becaused we are waiting for a slow transition CB")
|
|
else:
|
|
if self.open_position == 100:
|
|
self.log("Opening shutter")
|
|
log_evaluation_result(smart_conditions)
|
|
self.call_service("cover/open_cover", entity_id = self.entity_id)
|
|
else:
|
|
log_evaluation_result(smart_conditions)
|
|
self.log("Opening shutter at " + str(self.open_position) + " (previous position was " + str(current_position) + ")")
|
|
self.call_service("cover/set_cover_position", entity_id = self.entity_id,position = self.open_position)
|
|
|
|
else : self.run_in(self.on_retry_update,60) #during HASS restart, shutter can be irresponsive untill the z-wave network is properly initialized
|
|
else:
|
|
log_evaluation_result(smart_conditions)
|
|
self.stop_slow_transition()
|
|
|
|
|
|
|