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()