325 lines
12 KiB
Python
325 lines
12 KiB
Python
"""Define a generic object which all apps/automations inherit from."""
|
||
from typing import Callable, Dict, List, Union
|
||
|
||
import voluptuous as vol
|
||
from appdaemon.plugins.hass.hassapi import Hass # pylint: disable=no-name-in-module
|
||
|
||
from const import (
|
||
CONF_ENTITY_IDS,
|
||
CONF_PROPERTIES,
|
||
EVENT_MODE_CHANGE,
|
||
OPERATOR_ALL,
|
||
OPERATORS,
|
||
THRESHOLD_CLOUDY,
|
||
)
|
||
|
||
CONF_CLASS = "class"
|
||
CONF_MODULE = "module"
|
||
CONF_DEPENDENCIES = "dependencies"
|
||
|
||
CONF_APP = "app"
|
||
CONF_CONSTRAINTS = "constraints"
|
||
CONF_MODE_ALTERATIONS = "mode_alterations"
|
||
|
||
CONF_ENABLED_TOGGLE_ENTITY_ID = "enabled_toggle_entity_id"
|
||
CONF_INITIAL = "initial"
|
||
CONF_NAME = "name"
|
||
|
||
CONF_OPERATOR = "operator"
|
||
|
||
SENSOR_CLOUD_COVER = "sensor.dark_sky_cloud_coverage"
|
||
|
||
APP_SCHEMA = vol.Schema(
|
||
{
|
||
vol.Required(CONF_MODULE): str,
|
||
vol.Required(CONF_CLASS): str,
|
||
vol.Optional(CONF_APP): str,
|
||
vol.Optional(CONF_CONSTRAINTS): vol.Schema(
|
||
{
|
||
vol.Optional(CONF_OPERATOR, default=OPERATOR_ALL): vol.In(OPERATORS),
|
||
vol.Required(CONF_CONSTRAINTS): dict,
|
||
}
|
||
),
|
||
vol.Optional(CONF_ENABLED_TOGGLE_ENTITY_ID): str,
|
||
},
|
||
extra=vol.ALLOW_EXTRA,
|
||
)
|
||
|
||
|
||
class Base(Hass):
|
||
"""Define a base app/automation object."""
|
||
|
||
APP_SCHEMA = APP_SCHEMA
|
||
|
||
def initialize(self) -> None:
|
||
"""Initialize."""
|
||
try:
|
||
self.APP_SCHEMA(self.args)
|
||
except vol.Invalid as err:
|
||
self.error("Invalid app schema: {0}".format(err), level="ERROR")
|
||
return
|
||
|
||
# Define a holding place for HASS entity IDs:
|
||
self.entity_ids = self.args.get(CONF_ENTITY_IDS, {})
|
||
|
||
# Define a holding place for any scheduler handles that the app wants to keep
|
||
# track of:
|
||
self.handles = {} # type: Dict[str, Callable]
|
||
|
||
# Define a holding place for key/value properties for this app:
|
||
self.properties = self.args.get(CONF_PROPERTIES, {})
|
||
|
||
# Define a holding place for any mode alterations for this app:
|
||
self.mode_alterations = self.args.get(CONF_MODE_ALTERATIONS, [])
|
||
|
||
# Take every dependecy and create a reference to it:
|
||
for app in self.args.get(CONF_DEPENDENCIES, []):
|
||
if not getattr(self, app, None):
|
||
setattr(self, app, self.get_app(app))
|
||
|
||
# Define a reference to the "manager app" – for example, a trash-
|
||
# related app might carry a reference to TrashManager:
|
||
if self.args.get(CONF_APP):
|
||
self.app = getattr(self, self.args[CONF_APP])
|
||
|
||
# Set the entity ID of the input boolean that will control whether
|
||
# this app is enabled or not:
|
||
if self.args.get(CONF_ENABLED_TOGGLE_ENTITY_ID):
|
||
self._enabled_toggle_entity_id = self.args[CONF_ENABLED_TOGGLE_ENTITY_ID]
|
||
else:
|
||
self._enabled_toggle_entity_id = "input_boolean.{0}".format(self.name)
|
||
|
||
# Register custom constraints:
|
||
self.register_constraint("constrain_anyone")
|
||
self.register_constraint("constrain_cloudy")
|
||
self.register_constraint("constrain_enabled")
|
||
self.register_constraint("constrain_everyone")
|
||
self.register_constraint("constrain_in_blackout")
|
||
self.register_constraint("constrain_noone")
|
||
self.register_constraint("constrain_sun")
|
||
|
||
# Run any initial configuration:
|
||
if hasattr(self, "configure"):
|
||
self.configure()
|
||
|
||
# Set up connections to the automation being disabled/enabled:
|
||
if self._enabled_entity_exists():
|
||
# Listen and track mode changes so that the app can respond as needed:
|
||
self.mode_events = [] # type: List[str]
|
||
self.listen_event(self._on_mode_change, EVENT_MODE_CHANGE)
|
||
|
||
# If the app has defined callbacks fror when the app is enabled or disabled,
|
||
# attach them to listeners. Note that we utilize `_on_*` here
|
||
# (leading underscore) – we do this so automations don't have to remember
|
||
# callback method signatures:
|
||
if hasattr(self, "on_disable"):
|
||
super().listen_state(
|
||
self._on_disable, self._enabled_toggle_entity_id, new="off"
|
||
)
|
||
if hasattr(self, "on_enable"):
|
||
super().listen_state(
|
||
self._on_enable, self._enabled_toggle_entity_id, new="on"
|
||
)
|
||
|
||
@property
|
||
def enabled(self) -> bool:
|
||
"""Return whether the app is enabled."""
|
||
if not self._enabled_entity_exists():
|
||
return True
|
||
return self.get_state(self._enabled_toggle_entity_id) == "on"
|
||
|
||
def _attach_constraints(
|
||
self, method: Callable, callback: Callable, *args: list, **kwargs: dict
|
||
) -> Union[str, list]:
|
||
"""Attach the constraint mechanism to an AppDaemon listener."""
|
||
if not self.args.get(CONF_CONSTRAINTS):
|
||
return method(callback, *args, **kwargs)
|
||
|
||
constraints = self.args[CONF_CONSTRAINTS][CONF_CONSTRAINTS]
|
||
|
||
if self.args[CONF_CONSTRAINTS].get(CONF_OPERATOR) == OPERATOR_ALL:
|
||
return method(callback, *args, **constraints, **kwargs)
|
||
|
||
handles = [] # type: List[str]
|
||
for name, value in constraints.items():
|
||
method(callback, *args, **{name: value}, **kwargs)
|
||
return handles
|
||
|
||
def _constrain_presence(self, method: str, value: Union[str, None]) -> bool:
|
||
"""Constrain presence in a generic fashion."""
|
||
if not value:
|
||
return True
|
||
|
||
return getattr(self.presence_manager, method)(
|
||
*[self.presence_manager.HomeStates[s] for s in value.split(",")]
|
||
)
|
||
|
||
def _enabled_entity_exists(self) -> bool:
|
||
"""Return True if the enabled entity exists."""
|
||
return self.entity_exists(self._enabled_toggle_entity_id)
|
||
|
||
def _on_disable(
|
||
self, entity: Union[str, dict], attribute: str, old: str, new: str, kwargs: dict
|
||
) -> None:
|
||
"""Set a listener for when the automation is disabled."""
|
||
self.on_disable()
|
||
|
||
def _on_enable(
|
||
self, entity: Union[str, dict], attribute: str, old: str, new: str, kwargs: dict
|
||
) -> None:
|
||
"""Set a listener for when the automation is enabled."""
|
||
self.on_enable()
|
||
|
||
def _on_mode_change(self, event_name: str, data: dict, kwargs: dict) -> None:
|
||
"""Compare mode changes to registered mode alterations."""
|
||
mode = data["name"]
|
||
|
||
if data["state"] == "on":
|
||
self.mode_events.append(mode)
|
||
elif mode in self.mode_events:
|
||
self.mode_events.remove(mode)
|
||
|
||
try:
|
||
primary = max(
|
||
(m for m in self.mode_alterations if m["mode"] in self.mode_events),
|
||
key=lambda m: m["priority"],
|
||
)
|
||
except ValueError:
|
||
try:
|
||
primary = next((m for m in self.mode_alterations if m["mode"] == mode))
|
||
except StopIteration:
|
||
return
|
||
|
||
if primary["action"] == "enable":
|
||
primary["action"] = "disable"
|
||
else:
|
||
primary["action"] = "enable"
|
||
|
||
# If the primary mode alteration prescribes an action that matches the state the
|
||
# app is already in, return:
|
||
if (self.enabled and primary["action"] == "enable") or (
|
||
not self.enabled and primary["action"] == "disable"
|
||
):
|
||
return
|
||
|
||
if primary["action"] == "enable":
|
||
self.enable()
|
||
else:
|
||
self.disable()
|
||
|
||
def constrain_anyone(self, value: str) -> bool:
|
||
"""Constrain execution to whether anyone is in a state."""
|
||
return self._constrain_presence("anyone", value)
|
||
|
||
def constrain_enabled(self, value: bool) -> bool:
|
||
"""Constrain execution to whether anyone is in a state."""
|
||
if value:
|
||
return self.enabled
|
||
return False
|
||
|
||
def constrain_cloudy(self, value: bool) -> bool:
|
||
"""Constrain execution based whether it's cloudy or not."""
|
||
cloud_cover = float(self.get_state(SENSOR_CLOUD_COVER))
|
||
if (value and cloud_cover >= THRESHOLD_CLOUDY) or (
|
||
not value and cloud_cover < THRESHOLD_CLOUDY
|
||
):
|
||
return True
|
||
return False
|
||
|
||
def constrain_everyone(self, value: str) -> bool:
|
||
"""Constrain execution to whether everyone is in a state."""
|
||
return self._constrain_presence("everyone", value)
|
||
|
||
def constrain_in_blackout(self, state: str) -> bool:
|
||
"""Constrain execution based on blackout state."""
|
||
if state is True:
|
||
return self.get_state("input_boolean.blackout_mode") == "on"
|
||
|
||
return self.get_state("input_boolean.blackout_mode") == "off"
|
||
|
||
def constrain_noone(self, value: str) -> bool:
|
||
"""Constrain execution to whether no one is in a state."""
|
||
return self._constrain_presence("noone", value)
|
||
|
||
def constrain_sun(self, position: str) -> bool:
|
||
"""Constrain execution to the location of the sun."""
|
||
if (position == "up" and self.sun_up()) or (
|
||
position == "down" and self.sun_down()
|
||
):
|
||
return True
|
||
return False
|
||
|
||
def disable(self) -> None:
|
||
"""Disable the app."""
|
||
if not self.entity_exists(self._enabled_toggle_entity_id):
|
||
return
|
||
|
||
self.turn_off(self._enabled_toggle_entity_id)
|
||
|
||
def enable(self) -> None:
|
||
"""Enable the app."""
|
||
if not self.entity_exists(self._enabled_toggle_entity_id):
|
||
return
|
||
|
||
self.turn_on(self._enabled_toggle_entity_id)
|
||
|
||
def listen_ios_event(self, callback: Callable, action: str) -> None:
|
||
"""Register a callback for an iOS event."""
|
||
self.listen_event(
|
||
callback,
|
||
"ios.notification_action_fired",
|
||
actionName=action,
|
||
constrain_enabled=True,
|
||
)
|
||
|
||
def listen_event(self, callback, event=None, auto_constraints=False, **kwargs):
|
||
"""Wrap AppDaemon's `listen_event` with the constraint mechanism."""
|
||
if not auto_constraints:
|
||
return super().listen_event(callback, event, **kwargs)
|
||
|
||
return self._attach_constraints(super().listen_event, callback, event, **kwargs)
|
||
|
||
def listen_state(self, callback, entity=None, auto_constraints=False, **kwargs):
|
||
"""Wrap AppDaemon's `listen_state` with the constraint mechanism."""
|
||
if not auto_constraints:
|
||
return super().listen_state(callback, entity, **kwargs)
|
||
|
||
return self._attach_constraints(
|
||
super().listen_state, callback, entity, **kwargs
|
||
)
|
||
|
||
def run_daily(self, callback, start, auto_constraints=False, **kwargs):
|
||
"""Wrap AppDaemon's `run_daily` with the constraint mechanism."""
|
||
if not auto_constraints:
|
||
return super().run_daily(callback, start, **kwargs)
|
||
|
||
return self._attach_constraints(super().run_daily, callback, start, **kwargs)
|
||
|
||
def run_at_sunrise(self, callback, *args, auto_constraints=False, **kwargs):
|
||
"""Wrap AppDaemon's `run_at_sunrise` with the constraint mechanism."""
|
||
if not auto_constraints:
|
||
return super().run_at_sunrise(callback, **kwargs)
|
||
|
||
return self._attach_constraints(super().run_at_sunrise, callback, **kwargs)
|
||
|
||
def run_at_sunset(
|
||
self,
|
||
callback: Callable[..., None],
|
||
*args: list,
|
||
auto_constraints=False,
|
||
**kwargs: dict
|
||
):
|
||
"""Wrap AppDaemon's `run_at_sunset` with the constraint mechanism."""
|
||
if not auto_constraints:
|
||
return super().run_at_sunset(callback, **kwargs)
|
||
|
||
return self._attach_constraints(super().run_at_sunset, callback, **kwargs)
|
||
|
||
def run_every(self, callback, start, interval, auto_constraints=False, **kwargs):
|
||
"""Wrap AppDaemon's `run_every` with the constraint mechanism."""
|
||
if not auto_constraints:
|
||
return super().run_every(callback, start, interval, **kwargs)
|
||
|
||
return self._attach_constraints(
|
||
super().run_every, callback, start, interval, **kwargs
|
||
) |