How to use appdaemon - 10 common examples

import logging import warnings import appdaemon.utils import mock from appdaemon.plugins.hass.hassapi import Hass from packaging.version import Version CURRENT_APPDAEMON_VERSION = Version(appdaemon.utils.__version__) def is_appdaemon_version_at_least(version_as_string): expected_appdaemon_version = Version(version_as_string) return CURRENT_APPDAEMON_VERSION >= expected_appdaemon_version class _DeprecatedAppdaemonVersionWarning: already_warned_during_this_test_session = False min_supported_appdaemon_version = '4.0.0'  @classmethod def show_warning_only_once(cls): if cls.already_warned_during_this_test_session is True: return cls.already_warned_during_this_test_session = True

import appdaemon.plugins.hass.hassapi as hass """ Notifies the specified notifier on the days you specify. """ class Notify(hass.Hass): def initialize(self): if "countdown_entity" in self.args: self.listen_state(self.evaluate_notice, self.args["countdown_entity"]) def evaluate_notice(self, entity, attribute, old, new, kwargs): days_remaining = self.get_state(entity, attribute="days_remaining") entity_friendly_name = self.get_state(entity, attribute="friendly_name") if type(self.args["notification_days"]) == int: notification_days = [self.args["notification_days"]] else: notification_days = [int(day) for day in self.args["notification_days"].split(",")] if days_remaining in notification_days: self.send_notice() elif self.args["notify_overdue"] and days_remaining < 0:

def check_constraint(key, value): unconstrained = True  with conf.ha_state_lock: if key == "constrain_input_boolean": values = value.split(",") if len(values) == 2: entity = values[0] state = values[1] else: entity = value state = "on" if entity in conf.ha_state and conf.ha_state[entity]["state"] != state: unconstrained = False if key == "constrain_input_select": values = value.split(",") entity = values.pop(0) if entity in conf.ha_state and conf.ha_state[entity]["state"] not in values: unconstrained = False if key == "constrain_presence":

raise ValueError("Invalid scheme for 'dash_url' - only HTTP is supported") dash_net = url.netloc.split(":") conf.dash_host = dash_net[0] try: conf.dash_port = dash_net[1] except IndexError: conf.dash_port = 80 if conf.dash_host == "": raise ValueError("Invalid host for 'dash_url'") if conf.threads is None: conf.threads = 10  if conf.logfile is None: conf.logfile = "STDOUT" if conf.errorfile is None: conf.errorfile = "STDERR" if isdaemon and ( conf.logfile == "STDOUT" or conf.errorfile == "STDERR" or conf.logfile == "STDERR" or conf.errorfile == "STDOUT" ): raise ValueError("STDOUT and STDERR not allowed with -d") # Setup Logging conf.logger = logging.getLogger("log1") numeric_level = getattr(logging, args.debug, None) conf.logger.setLevel(numeric_level)

conf.app_dir = config['AppDaemon'].get("app_dir") conf.dashboard_dir = config['AppDaemon'].get("dash_dir") conf.timeout = config['AppDaemon'].get("timeout") if config['AppDaemon'].get("disable_apps") == "1": conf.apps = False else: conf.apps = True if config['AppDaemon'].get("dash_force_compile") == "1": conf.dash_force_compile = True else: conf.dash_force_compile = False if config['AppDaemon'].get("dash_compile_on_start") == "1":  conf.dash_compile_on_start = True else: conf.dash_compile_on_start = False if conf.dash_url is not None: conf.dashboard = True url = urlparse(conf.dash_url) if url.scheme != "http": raise ValueError("Invalid scheme for 'dash_url' - only HTTP is supported") dash_net = url.netloc.split(":") conf.dash_host = dash_net[0] try: conf.dash_port = dash_net[1] except IndexError: conf.dash_port = 80

# Get command line args parser = argparse.ArgumentParser() parser.add_argument("-c", "--config", help="full path to config directory", type=str, default=None) parser.add_argument("-p", "--pidfile", help="full path to PID File", default=None) parser.add_argument("-t", "--timewarp", help="speed that the scheduler will work at for time travel", default=1, type=float) parser.add_argument("-s", "--starttime", help="start time for scheduler ", type=str) parser.add_argument("-e", "--endtime", help="end time for scheduler ", type=str, default=None) parser.add_argument("-D", "--debug", help="global debug level", default="INFO", choices= [ "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL" ]) parser.add_argument('-m', '--moduledebug', nargs=2, action='append', help=argparse.SUPPRESS)  parser.add_argument('-v', '--version', action='version', version='%(prog)s ' + utils.__version__) parser.add_argument('--profiledash', help=argparse.SUPPRESS, action='store_true') args = parser.parse_args() config_dir = args.config pidfile = args.pidfile if config_dir is None: config_file_yaml = utils.find_path("appdaemon.yaml") else: config_file_yaml = os.path.join(config_dir, "appdaemon.yaml") if config_file_yaml is None: print("FATAL: no configuration directory defined and defaults not present\n") parser.print_help() sys.exit(1)

init = getattr(self.objects[name]["object"], "initialize", None) if init is None: self.logger.warning("Unable to find initialize() function in module %s - skipped", name) await self.increase_inactive_apps(name) return else: self.logger.warning("Unable to find module %s - initialize() skipped", name) await self.increase_inactive_apps(name) return # Call its initialize function try: if asyncio.iscoroutinefunction(init): await init() else:  await utils.run_in_executor(self, init) await self.set_state(name, state="idle") await self.increase_active_apps(name) event_data = { "event_type": "app_initialized", "data": {'app' : name} } await self.AD.events.process_event("admin", event_data) except TypeError: self.AD.threading.report_callback_sig(name, "initialize", init, {}) except: error_logger = logging.getLogger("Error.{}".format(name)) error_logger.warning('-' * 60) error_logger.warning("Unexpected error running initialize() for %s", name)

async def check_days_constraint(self, args, name): unconstrained = True if "constrain_days" in args: days = args["constrain_days"] now = await self.AD.sched.get_now() daylist = [] for day in days.split(","):  daylist.append(await utils.run_in_executor(self, utils.day_of_week, day)) if now.weekday() not in daylist: unconstrained = False return unconstrained

def get_state(self, entity_id=None, attribute=None):  utils.log(conf.logger, "DEBUG", "get_state: {}.{}".format(entity_id, attribute)) device = None entity = None if entity_id is not None and "." in entity_id: if not self.entity_exists(entity_id): return None if entity_id is not None: if "." not in entity_id: if attribute is not None: raise ValueError( "{}: Invalid entity ID: {}".format(self.name, entity)) device = entity_id entity = None else: device, entity = entity_id.split(".") with conf.ha_state_lock:

Harcoded custom logic for controlling HA with feedback from these actions. """ from datetime import datetime as dt from dateutil.parser import parse import appdaemon.appapi as appapi import appdaemon.conf as conf # DELAY_TO_SET_DEFAULT_TARGET = 1800 # sec DELAY_TO_SET_DEFAULT_TARGET = 120 # sec # noinspection PyClassHasNoInit class FamilyTracker(appapi.AppDaemon): """Family Tracker.""" _tracking_state = None _telegram_targets = None _notifier = None _timer_update_target = None _base_url = None _anybody_home = None def initialize(self): """AppDaemon required method for app init.""" config = dict(self.config['AppDaemon']) _chatids = [int(x) for x in config.get('bot_chatids').split(',')] self._notifier = config.get('notifier').replace('.', '/') self._base_url = config.get('base_url').replace('.', '/') self._anybody_home = False