Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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
# 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)
sys.exit(1)
if "logs" in config:
logs = config["logs"]
else:
logs = {}
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")
print("ERROR", "'log' directive deprecated, please convert to new 'logs' syntax")
sys.exit(1)
if "logs" in config:
logs = config["logs"]
else:
logs = {}
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"))
exit = False
if "time_zone" not in config["appdaemon"]:
self.logger.error("time_zone not specified in appdaemon.cfg")
exit = True
if "latitude" not in config["appdaemon"]:
self.logger.error("latitude not specified in appdaemon.cfg")
exit = True
if "longitude" not in config["appdaemon"]:
def get_ad_version():
"""Returns a string with the current version of AppDaemon.
Examples:
>>> version = self.get_ad_version()
"""
return utils.__version__
import appdaemon.events as events
import appdaemon.services as services
import appdaemon.sequences as sequences
import appdaemon.scheduler as scheduler
self.logging = logging
self.logging.register_ad(self)
self.logger = logging.get_logger()
self.threading = None
self.callbacks = None
self.futures = None
self.state = None
self.config = kwargs
self.booted = "booting"
self.config["ad_version"] = utils.__version__
self.check_app_updates_profile = ""
self.was_dst = False
self.last_state = None
self.executor = None
self.loop = None
self.srv = None
self.appd = None
self.stopping = False
self.http = None
self.admin_loop = None
self.global_vars = {}
self.global_lock = threading.RLock()
self.AD.loop.create_task(self.AD.sched.loop())
if self.AD.apps is True:
self.logger.debug("Reading Apps")
await self.AD.app_management.check_app_updates()
self.logger.info("App initialization complete")
#
# Fire APPD Started Event
#
await self.AD.events.process_event("global", {"event_type": "appd_started", "data": {}})
self.booted = await self.AD.sched.get_now()
await self.AD.state.add_entity("admin", "sensor.appdaemon_version", utils.__version__)
await self.AD.state.add_entity("admin", "sensor.appdaemon_uptime", str(datetime.timedelta(0)))
await self.AD.state.add_entity("admin", "sensor.appdaemon_booted", utils.dt_to_str((await self.AD.sched.get_now()).replace(microsecond=0), self.AD.tz))
warning_step = 0
warning_iterations = 0
s1 = 0
e1 = 0
# Start the loop proper
while not self.stopping:
start_time = datetime.datetime.now().timestamp()
try:
if self.AD.apps is True:
# 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="/tmp/hapush.pid")
parser.add_argument("-t", "--tick", help="time that a tick in the schedular lasts (seconds)", 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("-i", "--interval", help="multiplier for scheduler tick", type=float, default=None)
parser.add_argument("-D", "--debug", help="debug level", default="INFO", choices=
[
"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"
])
parser.add_argument('-v', '--version', action='version', version='%(prog)s ' + utils.__version__)
parser.add_argument('--profiledash', help=argparse.SUPPRESS, action='store_true')
# Windows does not have Daemonize package so disallow
if platform.system() != "Windows":
parser.add_argument("-d", "--daemon", help="run as a background process", action="store_true")
args = parser.parse_args()
config_dir = args.config
if platform.system() != "Windows":
from daemonize import Daemonize
if platform.system() != "Windows":
isdaemon = args.daemon
else:
async def hello(self, data):
if "client_name" not in data:
raise RequestHandlerException('client_name required')
if self.AD.http.password is None:
self.authed = True
if not self.authed:
await self._auth_data(data)
if not self.authed:
raise RequestHandlerException('authorization failed')
self.access.info("New client %s connected", data["client_name"])
response_data = {
"version": utils.__version__
}
return response_data
self.access.setLevel(numeric_level)
self.access.propagate = False
# formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
efh = RotatingFileHandler(
config['log'].get("accessfile"), maxBytes=log_size, backupCount=log_generations
)
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)