Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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
if config_dir is None:
config_file = find_path("appdaemon.cfg")
else:
config_file = os.path.join(config_dir, "appdaemon.cfg")
if platform.system() != "Windows":
isdaemon = args.daemon
else:
isdaemon = False
# Read Config File
config = configparser.ConfigParser()
config.read_file(open(config_file))
conf.config_dir = os.path.dirname(config_file)
assert "AppDaemon" in config, "[AppDaemon] section required in {}".format(
config_file
)
conf.config = config
conf.ha_url = config['AppDaemon']['ha_url']
conf.ha_key = config['AppDaemon'].get('ha_key', "")
conf.logfile = config['AppDaemon'].get("logfile")
conf.errorfile = config['AppDaemon'].get("errorfile")
conf.app_dir = config['AppDaemon'].get("app_dir")
conf.threads = int(config['AppDaemon'].get('threads'))
conf.certpath = config['AppDaemon'].get("cert_path")
conf.dash_url = config['AppDaemon'].get("dash_url")
conf.app_dir = config['AppDaemon'].get("app_dir")
conf.dashboard_dir = config['AppDaemon'].get("dash_dir")
mod = -1
while True:
try:
next_rising_dt = conf.location.sunrise(
now + datetime.timedelta(days=mod), local=False
)
if next_rising_dt > now:
break
except astral.AstralError:
pass
mod += 1
mod = -1
while True:
try:
next_setting_dt = conf.location.sunset(
now + datetime.timedelta(days=mod), local=False
)
if next_setting_dt > now:
break
except astral.AstralError:
pass
mod += 1
old_next_rising_dt = conf.sun.get("next_rising")
old_next_setting_dt = conf.sun.get("next_setting")
conf.sun["next_rising"] = next_rising_dt
conf.sun["next_setting"] = next_setting_dt
if old_next_rising_dt is not None \
and old_next_rising_dt != conf.sun["next_rising"]:
# dump_schedule()
mod = -1
while True:
try:
next_setting_dt = conf.location.sunset(
now + datetime.timedelta(days=mod), local=False
)
if next_setting_dt > now:
break
except astral.AstralError:
pass
mod += 1
old_next_rising_dt = conf.sun.get("next_rising")
old_next_setting_dt = conf.sun.get("next_setting")
conf.sun["next_rising"] = next_rising_dt
conf.sun["next_setting"] = next_setting_dt
if old_next_rising_dt is not None \
and old_next_rising_dt != conf.sun["next_rising"]:
# dump_schedule()
process_sun("next_rising")
# dump_schedule()
if old_next_setting_dt is not None \
and old_next_setting_dt != conf.sun["next_setting"]:
# dump_schedule()
process_sun("next_setting")
# dump_schedule()
def get_callback_entries(self):
callbacks = {}
for name in conf.callbacks.keys():
callbacks[name] = {}
utils.log(conf.logger, "INFO", "{}:".format(name))
for uuid_ in conf.callbacks[name]:
callbacks[name][uuid_] = {}
if "entity" in callbacks[name][uuid_]:
callbacks[name][uuid_]["entity"] = conf.callbacks[name][uuid_]["entity"]
else:
callbacks[name][uuid_]["entity"] = None
callbacks[name][uuid_]["type"] = conf.callbacks[name][uuid_]["type"]
callbacks[name][uuid_]["kwargs"] = conf.callbacks[name][uuid_]["kwargs"]
callbacks[name][uuid_]["function"] = conf.callbacks[name][uuid_]["function"]
callbacks[name][uuid_]["name"] = conf.callbacks[name][uuid_]["name"]
return(callbacks)
# Load apps
if conf.apps is True:
read_apps(True)
ha.log(conf.logger, "INFO", "App initialization complete")
#
# Fire HA_STARTED and APPD_STARTED Events
#
if first_time is True:
process_event({"event_type": "appd_started", "data": {}})
first_time = False
else:
process_event({"event_type": "ha_started", "data": {}})
if conf.version < parse_version('0.34') or conf.commtype == "SSE":
#
# Older version of HA - connect using SSEClient
#
if conf.commtype == "SSE":
ha.log(conf.logger, "INFO", "Using SSE")
else:
ha.log(
conf.logger, "INFO",
"Home Assistant version < 0.34.0 - "
"falling back to SSE"
)
headers = {'x-ha-access': conf.ha_key}
if conf.timeout is None:
messages = SSEClient(
"{}/api/stream".format(conf.ha_url),
verify=False, headers=headers, retry=3000
conf.config_dir = os.path.dirname(config_file)
assert "AppDaemon" in config, "[AppDaemon] section required in {}".format(
config_file
)
conf.config = config
conf.ha_url = config['AppDaemon']['ha_url']
conf.ha_key = config['AppDaemon'].get('ha_key', "")
conf.logfile = config['AppDaemon'].get("logfile")
conf.errorfile = config['AppDaemon'].get("errorfile")
conf.app_dir = config['AppDaemon'].get("app_dir")
conf.threads = int(config['AppDaemon'].get('threads'))
conf.certpath = config['AppDaemon'].get("cert_path")
conf.dash_url = config['AppDaemon'].get("dash_url")
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