Source code for juham.app.japp

from typing import List
from juham.base.group import Group
from juham.base.object import Object
import importlib.metadata  # for plugin architecture


[docs] class JApp(Group): """Default application class. Configures Base class with JPaho2 MQTT and JInflux time series database. It is up to the sub classes to initialize appropriate JInflux toke, org, host and database attributes to match the Influx account in question. """ _class_id = None plugins: List = [] def __init__(self, name: str) -> None: """Instantiates and initializes the Juham application with the given name using the default JPaho2 MQTT and JInflux time series database. It is up to the subclasses to initialize the appropriate MQTT and JInflux attributes, such as token, org, and host. The host and port class attributes for MQTT client is set to 'localhost' and 1883. The default logger instantiated with the application is configured to write log messages to both a log file and stdout. By default, the application log filename is set to the same as the application name. Args: name (str): The name of the application, determining the default log filename. """ super().__init__(name) self.children = [] self.configure()
[docs] @classmethod def load_plugins(cls) -> None: """Loads all plugins.""" cls.plugins = [] for entry_point in importlib.metadata.entry_points(group="juham.plugins"): plugin_class = entry_point.load() # Load the plugin class cls.plugins.append(plugin_class)
[docs] @classmethod def instantiate_plugins(self, plugin_classes: list) -> list: """Instantiates all installed plugin classes with their default names. This method is provided for testing purposes only, or for small home automation applications that needs only one instance per class. Args: plugins (list) : list of plugin classes Returns: list of plugin objects """ plugins = [] for plugin_class in plugin_classes: plugin_instance = plugin_class() # Instantiate the plugin class plugins.append(plugin_instance) # Append the instance to the plugins list return plugins
[docs] def add_all(self, plugins) -> None: """Add the plugin instances into the application.""" for plugin in plugins: # check if the plugin is already in the list if isinstance(plugin, Object): if self.find_plugin(plugin.name, type(plugin)) == None: self.add(plugin) self.info(f"Plugin {plugin.name}:{str(type(plugin))} installed") else: self.error( f"Plugin {plugin.name}:{str(type(plugin))} already installed" ) else: self.error(f"{type(plugin)} is not a Juham object")
[docs] def configure(cls) -> None: """Application instance specific configuration tasks. Up to the sub classes to implement. """
[docs] def find_plugin(self, name, cls): """Check if the application contains the given plugin, as defined by its name and class, and return the plugin. Returns: plugin object if found, None if the application does not contain a plugin of that name and type """ for p in self.children: if p.name == name and type(p) == cls: return p return None
[docs] @classmethod def register(cls): if cls.get_class_id == None: Group.register() cls.initialize_class()