import ad_toolbox.smartcondition as SmartCondition import ad_toolbox.smartvalue as SmartValue from ad_toolbox.eventhandler import EventHandler from ad_toolbox.smartobject import SmartObject import time class SmartHeatpump(SmartObject): AUTO_CONTROL_MODE = 'auto' FORCE_ON_CONTROL_MODE = 'on' FORCE_OFF_CONTROL_MODE = 'off' def on_initialize_smart_object(self): super().on_initialize_smart_object() self.event_handlers = [] self.control_mode_entity_id = None self.control_mode = self.AUTO_CONTROL_MODE if 'quick_resume' in self.args: self.quick_resume_sensor = self.create_entity(self.args['quick_resume']['sensor']) is_on = 'on' if self.entity.get_state() != 'off' else 'off' if self.get_state(self.args['quick_resume']['sensor']) != 'on': # if the sensor doesn't exist I wan't to set it as well self.set_state(self.args['quick_resume']['sensor'],state = is_on) self.register_reset_events(self.args['quick_resume'], "quick_resume") self.target_temperature = SmartValue.Evaluator(self,self.args['target_temperature'], expression_name = 'target_temperature', on_change_cb = self.on_change_target_temperature, constants = self.constants) self.hvac_mode = SmartValue.Evaluator(self,self.args['hvac_mode'], expression_name = 'hvac_mode', on_change_cb = self.on_change_hvac_mode, constants = self.constants) if 'horizontal_vane' in self.args: self.horizontal_vane = SmartValue.Evaluator(self,self.args['horizontal_vane'], expression_name = 'horizontal_vane', on_change_cb = self.on_change_horizontal_vane, constants = self.constants) else: self.horizontal_vane = None if 'fan_mode' in self.args: self.fan_mode = SmartValue.Evaluator(self,self.args['fan_mode'], expression_name = 'fan_mode', on_change_cb = self.on_change_fan_mode, constants = self.constants) else: self.fan_mode = None self.smart_conditions_evaluator = SmartCondition.Evaluator(self,self.args['smart_conditions'],condition_name = "smart_conditions",on_change_cb = self.on_smart_conditions_change,constants = self.constants, templates_library = self.templates_library) if 'control_mode_selector' in self.args or 'heatpump_override' in self.args: self.initialize_control_mode_selector() if 'input_buttons' in self.args: if 'turn_on' in self.args['input_buttons']: self.listen_state(lambda entity, attribute, old, new, kwargs: self.on_turn_on_button(),self.args['input_buttons']['turn_on']) if 'turn_off' in self.args['input_buttons']: self.listen_state(lambda entity, attribute, old, new, kwargs: self.on_turn_off_button(),self.args['input_buttons']['turn_off']) def initialize_control_mode_selector(self): selector_args = self.args['control_mode_selector'] if 'control_mode_selector' in self.args else self.args['heatpump_override'] if isinstance(selector_args, str): self.control_mode_entity_id = selector_args elif isinstance(selector_args, dict) and 'sensor' in selector_args: self.control_mode_entity_id = selector_args['sensor'] else: self.log_warning("control_mode_selector must be a string entity_id or a dict with a 'sensor' key") return self.create_entity( self.control_mode_entity_id, state = self.AUTO_CONTROL_MODE, attributes = { 'options': [self.AUTO_CONTROL_MODE, self.FORCE_ON_CONTROL_MODE, self.FORCE_OFF_CONTROL_MODE] }, icon = 'mdi:power-settings' ) current_mode = self.get_state(self.control_mode_entity_id) self.control_mode = self.normalize_control_mode(current_mode) self.listen_state(self.on_control_mode_change, self.control_mode_entity_id) self.listen_event( self.on_control_mode_select_option_service, 'call_service', domain='input_select', service='select_option', entity_id=self.control_mode_entity_id ) self.register_reset_events(selector_args, "control_mode_selector") self.apply_control_mode(self.control_mode) def on_control_mode_select_option_service(self, event_name, data, kwargs): service_data = data.get('service_data', {}) if isinstance(data, dict) else {} service_entity_id = service_data.get('entity_id') if isinstance(service_entity_id, list): if self.control_mode_entity_id not in service_entity_id: return elif isinstance(service_entity_id, str): if service_entity_id != self.control_mode_entity_id: return else: return mode = service_data.get('option') if mode == None or mode == self.control_mode: return self.set_state(self.control_mode_entity_id, state = mode) self.on_control_mode_change(self.control_mode_entity_id, 'state', self.control_mode, mode, {}) def register_reset_events(self, config, config_name): if not isinstance(config, dict): return reset_events = config.get('reset_events', config.get('reset_event')) if reset_events is None: return if isinstance(reset_events, str): reset_events = [reset_events] elif not isinstance(reset_events, list): self.log_warning(f"{config_name}.reset_events must be a string or a list of events") return self.event_handlers.append(EventHandler(self, reset_events, self.on_reset_event, config_name)) def on_reset_event(self, event_name, event_data, config_name): if config_name == "quick_resume": if 'quick_resume' in self.args: self.set_state(self.args['quick_resume']['sensor'], state='off') elif config_name == "control_mode_selector": if self.control_mode_entity_id: self.set_state(self.control_mode_entity_id, state=self.AUTO_CONTROL_MODE) def normalize_control_mode(self, mode): if mode in [self.AUTO_CONTROL_MODE, self.FORCE_ON_CONTROL_MODE, self.FORCE_OFF_CONTROL_MODE]: return mode return self.AUTO_CONTROL_MODE def on_control_mode_change(self, entity, attribute, old, new, kwargs): if old == new: return previous_mode = self.control_mode self.control_mode = self.normalize_control_mode(new) if self.control_mode != new: self.log_warning(f"Invalid control mode '{new}' on {entity}, falling back to '{self.control_mode}'") self.set_state(entity, state = self.control_mode) self.apply_control_mode(self.control_mode, previous_mode) def apply_control_mode(self, mode, previous_mode = None): if mode == self.AUTO_CONTROL_MODE: self.log_info("Heatpump control mode is auto, SmartCondition is enabled") self.smart_conditions_evaluator.activate() return self.smart_conditions_evaluator.deactivate() if mode == self.FORCE_ON_CONTROL_MODE: self.log_info("Heatpump control mode is on, forcing Heatpump on") self.start_cooling() elif mode == self.FORCE_OFF_CONTROL_MODE: self.log_info("Heatpump control mode is off, forcing Heatpump off") self.stop_cooling() if 'quick_resume' in self.args: self.set_state(self.args['quick_resume']['sensor'],state = 'off') def on_turn_on_button(self): self.start_cooling() def on_turn_off_button(self): self.stop_cooling() # we don't wan't quick resum if we force turn off the a/c if 'quick_resume' in self.args: self.set_state(self.args['quick_resume']['sensor'],state = 'off') def start_cooling(self): if 'quick_resume' in self.args: self.set_state(self.args['quick_resume']['sensor'],state = 'on') temperature = self.target_temperature.evaluate(False) hvac_mode = self.hvac_mode.evaluate(False) fan_mode = self.fan_mode.evaluate(False) if self.fan_mode else None self.log_info(f"Starting Heatpump in {hvac_mode} mode, with target_temperature of {temperature}° and {fan_mode} fan speed") #self.turn_on(self.entity_id) #time.sleep(2) self.call_service("climate/set_hvac_mode", entity_id = self.entity_id,hvac_mode = hvac_mode) time.sleep(1) self.call_service("climate/set_temperature", entity_id = self.entity_id,temperature = temperature,hvac_mode = hvac_mode) if self.fan_mode: time.sleep(1) self.call_service("climate/set_fan_mode", entity_id = self.entity_id,fan_mode = fan_mode) if self.horizontal_vane: time.sleep(1) horizontal_vane = self.horizontal_vane.evaluate(False) self.set_horizontal_vane(horizontal_vane) def stop_cooling(self): self.log_info("Setting Heatpump hvac_mode to off") self.call_service("climate/set_hvac_mode", entity_id = self.entity_id,hvac_mode = "off") def on_smart_conditions_change(self,prev_result,result): if self.control_mode != self.AUTO_CONTROL_MODE: self.log_info(f"SmartCondition update ignored because control mode is '{self.control_mode}'") return if result == SmartCondition.Result.Succeeded: self.start_cooling() else: self.log_info("Turning Off Heatpump") self.stop_cooling() def on_change_target_temperature(self,prev_result,result): if self.entity.get_state() != 'off': self.log_info(f"Setting target temperature to {result}°") self.call_service("climate/set_temperature", entity_id = self.entity_id,temperature = result) def on_change_hvac_mode(self,prev_result,result): if self.entity.get_state() != 'off': self.log_info(f"Setting hvac_mode to {result}") self.call_service("climate/set_hvac_mode", entity_id = self.entity_id,hvac_mode = result) def on_change_fan_mode(self,prev_result,result): if self.entity.get_state() != 'off': self.log_info(f"Setting fan_mode to {result}") self.call_service("climate/set_fan_mode", entity_id = self.entity_id,fan_mode = result) def set_horizontal_vane(self,value): device, entity = self.split_entity(self.entity_id) self.call_service("select/select_option", entity_id = f"select.{entity}_horizontal_vane",option = value) #self.set_state(f"select.{entity}_horizontal_vane",state = value) def on_change_horizontal_vane(self,prev_result,result): if self.entity.get_state() != 'off': self.log_info(f"Setting horizontal_vane to {result}") self.set_horizontal_vane(result)