From 92d7f1c869b1e5812da30d0666edcdf25179a7d4 Mon Sep 17 00:00:00 2001 From: Pierre Date: Mon, 15 Jun 2026 18:50:22 +0200 Subject: [PATCH] new smartshutter app --- apps/smartshutter.py | 247 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 247 insertions(+) create mode 100644 apps/smartshutter.py diff --git a/apps/smartshutter.py b/apps/smartshutter.py new file mode 100644 index 0000000..bfb3f51 --- /dev/null +++ b/apps/smartshutter.py @@ -0,0 +1,247 @@ +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() + + + \ No newline at end of file