144 lines
7.0 KiB
Python
144 lines
7.0 KiB
Python
import appdaemon.plugins.hass.hassapi as hass
|
|
import ad_toolbox.smartcondition as SmartCondition
|
|
from ad_toolbox.smartobject import SmartObject
|
|
|
|
class SmartSwitch(SmartObject):
|
|
|
|
#@SmartCondition.catch_smartcondition_exception(lambda self, message: self.log_error(message,stop_app = True))
|
|
def on_initialize_smart_object(self):
|
|
#super().initialize()
|
|
|
|
self.off_conditions_evaluator = None
|
|
self.smart_conditions_evaluator = None
|
|
|
|
#self.depends_on_module("smartswitch")
|
|
|
|
if "smart_conditions" in self.args:
|
|
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 "off_conditions" in self.args:
|
|
if self.smart_conditions_evaluator == None:
|
|
self.off_conditions_evaluator = SmartCondition.Evaluator(self,self.args['off_conditions'],condition_name = "off_conditions",on_succeed_cb = self.on_off_conditions,constants = self.constants, templates_library = self.templates_library)
|
|
else:
|
|
self.log(f"Warning you can't have both an off_conditons and a smart_conditions, the off_conditions will be ignored")
|
|
|
|
if "debug" in self.args:
|
|
self.log(f'Registering on_debug_display_event for {self.args["debug"]}')
|
|
self.listen_event(self.on_debug_display_event,self.args["debug"])
|
|
|
|
if "off_events" in self.args:
|
|
self.register_event_from_yaml(self.args["off_events"],self.on_turn_off_event)
|
|
if "on_events" in self.args:
|
|
self.register_event_from_yaml(self.args["on_events"],self.on_turn_on_event)
|
|
if "toggle_events" in self.args:
|
|
self.register_event_from_yaml(self.args["toggle_events"],self.on_toggle_event)
|
|
|
|
#todo: replace with register_event_from_yaml
|
|
if "toggle_action" in self.args:
|
|
self.toggle_action = self.args["toggle_action"]
|
|
self.listen_event(self.on_toggle_event_action,"ios.action_fired")
|
|
|
|
self.auto_switch_cb_handle = None
|
|
if 'auto_switch_on_after' in self.args:
|
|
if self.is_off():
|
|
self.log(f"Smartswitch has been restarted while it was on. Since we have auto_switch_on_after activated we turn it on as we can't know how long is left for the timer",level = 'WARNING')
|
|
self.switch_on()
|
|
self.listen_state(self.on_state_change,self.entity_id, old = 'on',new = 'off')
|
|
if 'auto_switch_off_after' in self.args:
|
|
if self.is_on():
|
|
self.log(f"Smartswitch has been restarted while it was on. Since we have auto_switch_off_after activated we turn it off as we can't know how long is left for the timer",level = 'WARNING')
|
|
self.switch_off()
|
|
self.listen_state(self.on_state_change,self.entity_id, old = 'off',new = 'on')
|
|
|
|
def terminate(self):
|
|
self.smart_conditions_evaluator = None
|
|
self.off_conditions_evaluator = None
|
|
super().terminate()
|
|
|
|
def on_state_change(self, entity, attribute, old, new, kwargs):
|
|
self.log("state changed from " + str(old) + " to " + str(new))
|
|
if old != new:
|
|
if self.auto_switch_cb_handle != None:
|
|
self.cancel_timer(self.auto_switch_cb_handle)
|
|
self.auto_switch_cb_handle = None
|
|
|
|
if new == 'on' and 'auto_switch_off_after' in self.args:
|
|
delay = self.args['auto_switch_off_after']
|
|
self.log(f"{self.entity_id} will auto switch on in {delay}s")
|
|
self.auto_switch_cb_handle = self.run_in(self.on_auto_switch_after, delay, new_state = 'off',autoswitch_delay = delay)
|
|
if new == 'off' and 'auto_switch_on_after' in self.args:
|
|
delay = self.args['auto_switch_on_after']
|
|
self.log(f"{self.entity_id} will auto switch off in {delay}s")
|
|
self.auto_switch_cb_handle = self.run_in(self.on_auto_switch_after, delay, new_state = 'on',autoswitch_delay = delay)
|
|
|
|
def on_auto_switch_after(self, kwargs):
|
|
self.auto_switch_cb_handle = None
|
|
self.log(f"Switching {self.entity_id} {kwargs['new_state']} after {kwargs['autoswitch_delay']}s")
|
|
self.set_state(self.entity_id,state = kwargs['new_state'])
|
|
|
|
#debug display to display all events
|
|
def on_debug_display_event(self,event_name,data,kwargs):
|
|
self.log(f"events {event_name} has been catched. data = {data}")
|
|
|
|
def switch_on(self):
|
|
self.log(f"Turn on {self.entity_id}")
|
|
self.turn_on(self.entity_id)
|
|
# not needed yet
|
|
# if self.smart_conditions_evaluator:
|
|
# self.smart_conditions_evaluator.force_last_evaluation_result(SmartCondition.Result.Succeeded)
|
|
# if self.off_conditions_evaluator:
|
|
# self.off_conditions_evaluator.force_last_evaluation_result(SmartCondition.Result.Succeeded)
|
|
|
|
def switch_off(self):
|
|
self.log(f"Turn off {self.entity_id}")
|
|
self.turn_off(self.entity_id)
|
|
# not needed yet
|
|
# if self.smart_conditions_evaluator:
|
|
# self.smart_conditions_evaluator.force_last_evaluation_result(SmartCondition.Result.Failed)
|
|
# if self.off_conditions_evaluator:
|
|
# self.off_conditions_evaluator.force_last_evaluation_result(SmartCondition.Result.Failed)
|
|
|
|
def on_toggle_event(self, event_name, data, kwargs):
|
|
if self.is_on():
|
|
self.log(f"Toggled off by event {event_name}")
|
|
self.switch_off()
|
|
else:
|
|
self.log(f"Toggled on by event {event_name}")
|
|
self.switch_on()
|
|
|
|
def is_on(self): return self.get_state(self.entity_id) != 'off'
|
|
|
|
def is_off(self): return not self.is_on()
|
|
|
|
def on_turn_off_event(self, event_name, data, kwargs):
|
|
self.log(f"Turned off by event {event_name}")
|
|
self.switch_off()
|
|
|
|
def on_turn_on_event(self, event_name, data, kwargs):
|
|
self.log(f"Turned on by event {event_name}")
|
|
self.switch_on()
|
|
|
|
def on_toggle_event_action(self, event_name, data, kwargs):
|
|
if data['actionName'] == self.toggle_action:
|
|
if self.is_on():
|
|
self.switch_off()
|
|
else:
|
|
self.switch_on()
|
|
|
|
def on_smart_conditions_change(self,prev_result,result):
|
|
#trying to track some weird behavior
|
|
# if self.smart_conditions_evaluator:
|
|
# debug_result = self.smart_conditions_evaluator.evaluate()
|
|
# if debug_result != result:
|
|
# self.log(f"on_smart_conditions_change was called with prev_result = {prev_result}, result = {result} and evaluate returned {debug_result}")
|
|
# else:
|
|
# self.log("on_smart_conditions_change was called without a smart_conditions_evaluator")
|
|
if result == SmartCondition.Result.Succeeded:
|
|
if self.is_off():
|
|
self.switch_on()
|
|
else :
|
|
if self.is_on():
|
|
self.switch_off()
|
|
|
|
def on_off_conditions(self):
|
|
if self.is_on():
|
|
self.switch_off() |