Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if "latitude" not in config["appdaemon"]:
self.logger.error("latitude not specified in appdaemon.cfg")
exit = True
if "longitude" not in config["appdaemon"]:
self.logger.error("longitude not specified in appdaemon.cfg")
exit = True
if "elevation" not in config["appdaemon"]:
self.logger.error("elevation not specified in appdaemon.cfg")
exit = True
if exit is True:
sys.exit(1)
utils.check_path("config_file", self.logger, config_file_yaml, pathtype="file")
if pidfile is not None:
self.logger.info("Using pidfile: %s", pidfile)
dir = os.path.dirname(pidfile)
name = os.path.basename(pidfile)
try:
with pid.PidFile(name, dir) as p:
self.run(appdaemon, hadashboard, admin, api, http)
except pid.PidFileError:
self.logger.error("Unable to aquire pidfile - terminating")
else:
self.run(appdaemon, hadashboard, admin, api, http)
)
efh.setLevel(numeric_level)
# efh.setFormatter(formatter)
self.access.addHandler(efh)
else:
self.access = self.logger
# Startup message
self.log(self.logger, "INFO", "AppDaemon Version {} starting".format(utils.__version__))
self.log(self.logger, "INFO", "Configuration read from: {}".format(config_file_yaml))
self.log(self.logger, "DEBUG", "AppDaemon Section: {}".format(config.get("AppDaemon")))
self.log(self.logger, "DEBUG", "HADashboard Section: {}".format(config.get("HADashboard")))
utils.check_path("config_file", self.logger, config_file_yaml, pathtype="file")
if isdaemon:
keep_fds = [fh.stream.fileno(), efh.stream.fileno()]
pid = args.pidfile
daemon = Daemonize(app="appdaemon", pid=pid, action=self.run,
keep_fds=keep_fds)
daemon.start()
while True:
time.sleep(1)
else:
self.run(appdaemon, hadashboard)
#
# Create some dirs
#
try:
js = os.path.join(self.compile_dir, "javascript")
css = os.path.join(self.compile_dir, "css")
if not os.path.isdir(self.compile_dir):
os.makedirs(self.compile_dir)
if not os.path.isdir(os.path.join(self.compile_dir, "javascript")):
os.makedirs(js)
if not os.path.isdir(os.path.join(self.compile_dir, "css")):
os.makedirs(css)
ha.check_path("css", self.logger, css, permissions="rwx")
ha.check_path("javascript", self.logger, js, permissions="rwx")
except:
self.logger.warning('-' * 60)
self.logger.warning("Unexpected error during HADashboard initialization")
self.logger.warning('-' * 60)
self.logger.warning(traceback.format_exc())
self.logger.warning('-' * 60)
#
# Set a start time
#
self.start_time = datetime.datetime.now()
# Create some dirs
#
try:
js = os.path.join(self.compile_dir, "javascript")
css = os.path.join(self.compile_dir, "css")
if not os.path.isdir(self.compile_dir):
os.makedirs(self.compile_dir)
if not os.path.isdir(os.path.join(self.compile_dir, "javascript")):
os.makedirs(js)
if not os.path.isdir(os.path.join(self.compile_dir, "css")):
os.makedirs(css)
ha.check_path("css", self.logger, css, permissions="rwx")
ha.check_path("javascript", self.logger, js, permissions="rwx")
except:
self.logger.warning('-' * 60)
self.logger.warning("Unexpected error during HADashboard initialization")
self.logger.warning('-' * 60)
self.logger.warning(traceback.format_exc())
self.logger.warning('-' * 60)
#
# Set a start time
#
self.start_time = datetime.datetime.now()
#
# Set up futures
#
self.futures = futures.Futures(self)
if self.apps is True:
if self.app_dir is None:
if self.config_dir is None:
self.app_dir = utils.find_path("apps")
self.config_dir = os.path.dirname(self.app_dir)
else:
self.app_dir = os.path.join(self.config_dir, "apps")
utils.check_path("config_dir", self.logger, self.config_dir, permissions="rwx")
utils.check_path("appdir", self.logger, self.app_dir)
# Initialize Apps
self.app_management = apps.AppManagement(self, kwargs.get("app_config_file", None))
# threading setup
self.threading = appdaemon.threading.Threading(self, kwargs)
self.stopping = False
#
# Set up Executor ThreadPool
#
if "threadpool_workers" in kwargs:
self.logging = logging.Logging(logs, args.debug)
self.logger = self.logging.get_logger()
if "time_zone" in config["appdaemon"]:
self.logging.set_tz(pytz.timezone(config["appdaemon"]["time_zone"]))
# Startup message
self.logger.info("AppDaemon Version %s starting", utils.__version__)
self.logger.info("Configuration read from: %s", config_file_yaml)
self.logging.dump_log_config()
self.logger.debug("AppDaemon Section: %s", config.get("appdaemon"))
self.logger.debug("HADashboard Section: %s", config.get("hadashboard"))
utils.check_path("config_file", self.logger, config_file_yaml, pathtype="file")
if pidfile is not None:
self.logger.info("Using pidfile: %s", pidfile)
dir = os.path.dirname(pidfile)
name = os.path.basename(pidfile)
try:
with pid.PidFile(name, dir) as p:
self.run(appdaemon, hadashboard, admin, api, http)
except pid.PidFileError:
self.logger.error("Unable to aquire pidfile - terminating")
else:
self.run(appdaemon, hadashboard, admin, api, http)
#
# Set up futures
#
self.futures = futures.Futures(self)
if self.apps is True:
if self.app_dir is None:
if self.config_dir is None:
self.app_dir = utils.find_path("apps")
self.config_dir = os.path.dirname(self.app_dir)
else:
self.app_dir = os.path.join(self.config_dir, "apps")
utils.check_path("config_dir", self.logger, self.config_dir, permissions="rwx")
utils.check_path("appdir", self.logger, self.app_dir)
# Initialize Apps
self.app_management = apps.AppManagement(self, kwargs.get("app_config_file", None))
# threading setup
self.threading = appdaemon.threading.Threading(self, kwargs)
self.stopping = False
#
# Set up Executor ThreadPool
#
if "threadpool_workers" in kwargs:
self.threadpool_workers = int(kwargs["threadpool_workers"])