from contextlib import ExitStack from dataclasses import dataclass, field from appdaemon.models.ad_config import AppDaemonConfig from context_manager import AppDaemonRunContext from subsystem import AppDaemon @dataclass class ADMain: """Class to contain the mechanics to run AppDaemon as module """ config_file: str cfg: AppDaemonConfig = field(init=False) _stack: ExitStack = field(default_factory=ExitStack) run_context: AppDaemonRunContext = field(init=False) ad: AppDaemon = field(init=False) def __post_init__(self) -> None: raw_cfg = read_config_file(self.config_file) self.cfg = AppDaemonConfig( config_file=self.config_file, **raw_cfg['appdaemon'] ) def __enter__(self): """Used to start the asyncio loop, thread pool, and AppDaemon""" # Use the ExitStack from ADMain instead of creating a new one self.run_context = AppDaemonRunContext(_stack=self._stack) self._stack.enter_context(self.run_context) # Start AppDaemon by entering it's context self.ad = AppDaemon(self.cfg, self.run_context) self._stack.enter_context(self.ad) return self def __exit__(self, exc_type, exc_value, traceback): self.ad.logger.debug(f'Closing context from {self.__class__.__name__}') self._stack.close() self.ad.logger.info('Stopped main()') def run(self): if hasattr(self, 'ad'): self.ad.logger.info('Running asyncio event loop indefinitely...') self.run_context.loop.run_forever() self.ad.logger.debug('Gracefully stopped event loop') else: logging.error('Running ADMain without context manager') if __name__ == '__main__': import logging.config from appdaemon.utils import read_config_file from rich.console import Console from rich.highlighter import NullHighlighter console = Console() logging.config.dictConfig( { 'version': 1, 'disable_existing_loggers': False, 'formatters': {'basic': {'style': '{', 'format': '[yellow]{name:20}[/] {message}'}}, 'handlers': { 'rich': { '()': 'rich.logging.RichHandler', 'formatter': 'basic', 'console': console, 'highlighter': NullHighlighter(), 'markup': True } }, 'root': {'level': 'DEBUG', 'handlers': ['rich']}, } ) with ADMain('/conf/ad-test/conf/appdaemon.yaml') as main: main.run() main.ad.logger.debug('Exiting main context from with statement')