230 lines
10 KiB
Python
230 lines
10 KiB
Python
import appdaemon.plugins.hass.hassapi as hass
|
|
import ad_toolbox.smartcondition as SmartCondition
|
|
from ad_toolbox.smartobject import SmartObject
|
|
from ad_toolbox.eventhandler import EventHandler
|
|
|
|
# =============================================================================
|
|
# SmartSwitch — Event- and condition-driven switch controller
|
|
# =============================================================================
|
|
# Binds a HA switch/input_boolean/light entity to a rich set of controls:
|
|
# HA events (on/off/toggle), smart conditions, and auto-timed switching.
|
|
#
|
|
# Inherits all SmartObject YAML keys (see smartobject.py).
|
|
# The "entity" key is required.
|
|
#
|
|
# YAML CONFIGURATION
|
|
# ------------------
|
|
#
|
|
# smart_conditions: # optional
|
|
# <SmartCondition block>
|
|
# When the condition result transitions to Succeeded the entity is
|
|
# turned on; when it transitions to Failed it is turned off.
|
|
# Mutually exclusive with off_conditions (off_conditions is ignored
|
|
# when both are set).
|
|
#
|
|
# off_conditions: # optional
|
|
# <SmartCondition block>
|
|
# Turns the entity off whenever the condition succeeds.
|
|
# Has no effect when smart_conditions is also configured.
|
|
#
|
|
# off_events: # optional
|
|
# <EventHandler events_block>
|
|
# HA events that turn the entity off.
|
|
#
|
|
# on_events: # optional
|
|
# <EventHandler events_block>
|
|
# HA events that turn the entity on.
|
|
#
|
|
# toggle_events: # optional
|
|
# <EventHandler events_block>
|
|
# HA events that toggle the entity.
|
|
#
|
|
# toggle_action: <ios_action_name> # optional
|
|
# iOS companion app action name (ios.action_fired) that toggles
|
|
# the entity.
|
|
#
|
|
# auto_switch_off_after: <seconds> # optional
|
|
# Automatically turn the entity off N seconds after it turns on.
|
|
# If the app restarts while the entity is already on the switch is
|
|
# turned off immediately (the remaining delay is not known).
|
|
#
|
|
# auto_switch_on_after: <seconds> # optional
|
|
# Automatically turn the entity on N seconds after it turns off.
|
|
# If the app restarts while the entity is already off the switch is
|
|
# turned on immediately.
|
|
#
|
|
# NOTES
|
|
# -----
|
|
# - smart_conditions and off_conditions use the SmartCondition evaluator;
|
|
# see smartcondition.py for the condition block schema.
|
|
# - off_events / on_events / toggle_events use the EventHandler format;
|
|
# see eventhandler.py for the events_block schema.
|
|
# - auto_switch_on_after and auto_switch_off_after are mutually independent
|
|
# and can coexist (e.g. pulse: on_after=0 + off_after=5).
|
|
#
|
|
# EXAMPLE YAML
|
|
# ------------
|
|
# my_switch:
|
|
# module: smartswitch
|
|
# class: SmartSwitch
|
|
# entity: input_boolean.my_switch
|
|
#
|
|
# on_events:
|
|
# btn:
|
|
# event_name: MY_BUTTON_PRESSED
|
|
# event_data:
|
|
# action: single
|
|
# off_events:
|
|
# btn_long:
|
|
# event_name: MY_BUTTON_PRESSED
|
|
# event_data:
|
|
# action: long
|
|
#
|
|
# auto_switch_off_after: 300
|
|
#
|
|
# smart_conditions:
|
|
# trigger:
|
|
# condition: sensor.presence == 'home'
|
|
# =============================================================================
|
|
|
|
class SmartSwitch(SmartObject):
|
|
|
|
def on_initialize_smart_object(self):
|
|
super().on_initialize_smart_object()
|
|
|
|
self.off_conditions_evaluator = None
|
|
self.smart_conditions_evaluator = None
|
|
|
|
# smart_conditions: auto on/off based on condition result
|
|
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)
|
|
|
|
# off_conditions: turn off only when condition succeeds (ignored if smart_conditions present)
|
|
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")
|
|
|
|
# Event-driven controls (off / on / toggle / iOS action)
|
|
self.event_handlers = []
|
|
if "off_events" in self.args:
|
|
self.event_handlers.append(EventHandler(self, self.args["off_events"], self.on_turn_off_event))
|
|
if "on_events" in self.args:
|
|
self.event_handlers.append(EventHandler(self, self.args["on_events"], self.on_turn_on_event))
|
|
if "toggle_events" in self.args:
|
|
self.event_handlers.append(EventHandler(self, self.args["toggle_events"], self.on_toggle_event))
|
|
|
|
# auto_switch_*_after: timed auto-switch; if the app restarts in the
|
|
# wrong state we apply the switch immediately since the delay is lost.
|
|
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()
|
|
|
|
# listen_state callback for auto_switch_*_after. Cancels any pending
|
|
# timer then arms a new one for the configured delay.
|
|
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)
|
|
|
|
# Timer callback: applies the deferred state set by auto_switch_*_after.
|
|
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'])
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public switch API
|
|
# ------------------------------------------------------------------
|
|
|
|
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()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Event callbacks
|
|
# ------------------------------------------------------------------
|
|
|
|
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()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Condition callbacks
|
|
# ------------------------------------------------------------------
|
|
|
|
# Fired by smart_conditions when its result changes. Mirrors the
|
|
# condition outcome directly onto the entity state.
|
|
def on_smart_conditions_change(self,prev_result,result):
|
|
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() |