237 lines
11 KiB
Python
237 lines
11 KiB
Python
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)
|
|
|
|
|
|
|
|
|