#!/usr/bin/env python3
# vim: set encoding=utf-8 tabstop=4 softtabstop=4 shiftwidth=4 expandtab
#########################################################################
# Copyright 2017- Martin Sinn m.sinn@gmx.de
#########################################################################
# This file is part of SmartHomeNG
#
# SmartHomeNG is free software: you can redistribute it and/or modifyNode.js Design Patterns - Second Edition
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# SmartHomeNG is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with SmartHomeNG If not, see <http://www.gnu.org/licenses/>.
##########################################################################
# TO DO
# - consolidate with plugin.py
"""
This library implements loading and starting of core modules of SmartHomeNG.
The methods of the class Modules implement the API for modules.
They can be used the following way: To call eg. **xxx()**, use the following syntax:
.. code-block:: python
from lib.module import Modules
modules = Modules.get_instance()
# to access a method (eg. enable_logic()):
moddules.xxx()
:Warning: This library is part of the core of SmartHomeNG. It **should not be called directly** from plugins!
"""
import logging
#import threading
import inspect
import os
import lib.config
import lib.translation as translation
from lib.constants import (KEY_CLASS_NAME, KEY_CLASS_PATH, KEY_INSTANCE,CONF_FILE)
from lib.utils import Utils
from lib.metadata import Metadata
logger = logging.getLogger(__name__)
_modules_instance = None # Pointer to the initialized instance of the Modules class (for use by static methods)
[Doku]class Modules():
"""
Module loader class. Parses config file and creates an instance for each module.
To start the modules, the start() method has to be called.
:param smarthome: Instance of the smarthome master-object
:param configfile: Basename of the module configuration file
:type samrthome: object
:type configfile: str
"""
_modules = []
_moduledict = {}
def __init__(self, smarthome, configfile):
self._sh = smarthome
self._basedir = smarthome.get_basedir()
# self._sh._moduledict = {}
global _modules_instance
if _modules_instance is not None:
import inspect
curframe = inspect.currentframe()
calframe = inspect.getouterframes(curframe, 4)
logger.critical("A second 'modules' object has been created. There should only be ONE instance of class 'Modules'!!! Called from: {} ({})".format(calframe[1][1], calframe[1][3]))
_modules_instance = self
# read module configuration (from etc/module.yaml)
_conf = lib.config.parse_basename(configfile, configtype='module')
if _conf == {}:
return
for module in _conf:
logger.debug("Modules, section: {}".format(module))
module_name, self.meta = self._get_modulename_and_metadata(module, _conf[module])
if module_name != '' and self.meta is not None:
if self.meta.test_shngcompatibility() and self.meta.test_pythoncompatibility():
args = self._get_conf_args(_conf[module])
classname, classpath = self._get_classname_and_classpath(_conf[module], module_name)
if not self._test_duplicate_configuration(module, classname):
try:
self._load_module(module, classname, classpath, args)
except Exception as e:
logger.exception("Module {0} exception: {1}".format(module, e))
else:
logger.warning("Section '{}' ignored".format(module))
logger.info('Loaded Modules: {}'.format( str( self.return_modules() ) ) )
# clean up (module configuration from module.yaml)
del(_conf) # clean up
return
def _get_modulename_and_metadata(self, module, mod_conf):
"""
Return the actual module name and the metadata instance
:param mod_conf: loaded section of the module.yaml for the actual module
:type mod_conf: dict
:return: module_name and metadata_instance
:rtype: string, object
"""
module_name = mod_conf.get('module_name','').lower()
meta = None
if module_name != '':
module_dir = os.path.join(self._basedir, 'modules', module_name)
if os.path.isdir(module_dir):
meta = Metadata(self._sh, module_name, 'module')
else:
logger.warning("Section '{}': No module directory {} found".format(module, module_dir))
else:
classpath = mod_conf.get(KEY_CLASS_PATH,'')
if classpath != '':
module_name = classpath.split('.')[len(classpath.split('.'))-1].lower()
logger.info("Section '{}': module_name '{}' was extracted from classpath '{}'".format(module, module_name, classpath))
meta = Metadata(self._sh, module_name, 'module', classpath)
else:
logger.info("Section '{}': No attribute 'module_name' found in configuration".format(module))
return (module_name, meta)
def _get_conf_args(self, mod_conf):
"""
Return the parameters/values for the actual module as args-dict
:param mod_conf: loaded section of the module.yaml for the actual module
:type mod_conf: dict
:return: args = specified parameters and their values
:rtype: dict
"""
args = {}
for arg in mod_conf:
if arg != KEY_CLASS_NAME and arg != KEY_CLASS_PATH and arg != KEY_INSTANCE:
value = mod_conf[arg]
if isinstance(value, str):
value = "'{0}'".format(value)
args[arg] = value
return args
def _get_classname_and_classpath(self, mod_conf, module_name):
"""
Returns the classname and the classpath for the actual module
:param mod_conf: loaded section of the module.yaml for the actual module
:param module_name: Module name (to be used, for building classpass, if it is not specified in the configuration
:type mod_conf: dict
:type module_name: str
:return: classname, classpass
:rtype: str, str
"""
classname = self.meta.get_string('classname')
if classname == '':
classname = mod_conf.get(KEY_CLASS_NAME,'')
try:
classpath = mod_conf[KEY_CLASS_PATH]
except:
classpath = 'modules.' + module_name
return (classname, classpath)
def _test_duplicate_configuration(self, module, classname):
"""
Returns True, if a module instance of the classname is already loaded by another configuration section
:param module: Name of the configuration
:param classname: Name of the class to check
:type module: str
:type classname: str
:return: True, if module is already loaded
:rtype: bool
"""
# give a warning if a module uses the same class twice
duplicate = False
for m in self._modules:
if m.__class__.__name__ == classname:
duplicate = True
logger.warning("Modules, section '{}': Multiple module instances of class '{}' detected, additional instance not initialized".format(module, classname))
return duplicate
def _load_module(self, name, classname, classpath, args):
"""
Module Loader. Loads one module defined by the parameters classname and classpath.
Parameters defined in the configuration file are passed to this function as 'args'
:param name: Section name in module configuration file (etc/module.yaml)
:param classname: Name of the (main) class in the module
:param classpath: Path to the Python file containing the class
:param args: Parameter as specified in the configuration file (etc/module.yaml)
:type name: str
:type classname: str
:type classpath: str
:type args: dict
:return: loaded module
:rtype: object
"""
logger.debug('_load_module: Section {}, Module {}, classpath {}'.format( name, classname, classpath ))
enabled = Utils.strip_quotes(args.get('enabled', 'true').lower())
if enabled == 'false':
logger.warning("Not loading module {} from section '{}': Module is disabled".format(classname, name))
return
logger.info("Loading module '{}': args = '{}'".format(name, args))
# Load an instance of the module
try:
exec("import {0}".format(classpath))
except Exception as e:
logger.critical("Module '{}' ({}) exception during import of __init__.py: {}".format(name, classpath, e))
return None
try:
exec("self.loadedmodule = {0}.{1}.__new__({0}.{1})".format(classpath, classname))
except Exception as e:
#logger.error("Module '{}' ({}) exception during initialization: {}".format(name, classpath, e))
pass
# load module-specific translations
translation.load_translations('module', classpath.replace('.', '/'), 'module/'+classpath.split('.')[1])
# translation.load_translations('global', classpath.replace('.', '/'), 'module/'+classpath.split('.')[1])
# get arguments defined in __init__ of module's class to self.args
try:
# exec("self.args = inspect.getargspec({0}.{1}.__init__)[0][1:]".format(classpath, classname))
exec("self.args = inspect.getfullargspec({0}.{1}.__init__)[0][1:]".format(classpath, classname))
except Exception as e:
logger.critical("Module '{}' ({}) exception during call to __init__.py: {}".format(name, classpath, e))
return None
#logger.warning("- self.args = '{}'".format(self.args))
# get list of argument used names, if they are defined in the module's class
logger.info("Module '{}': args = '{}'".format(classname, str(args)))
arglist = [name for name in self.args if name in args]
argstring = ",".join(["{}={}".format(name, args[name]) for name in arglist])
self.loadedmodule._init_complete = False
(module_params, params_ok, hide_params) = self.meta.check_parameters(args)
if params_ok == True:
if module_params != {}:
# initialize parameters the old way
argstring = ",".join(["{}={}".format(name, "'"+str(module_params.get(name,''))+"'") for name in arglist])
# initialize parameters the new way: Define a dict within the instance
self.loadedmodule._parameters = module_params
self.loadedmodule._metadata = self.meta
# initialize the loaded instance of the module
self.loadedmodule._init_complete = True # set to false by module, if an initalization error occurs
exec("self.loadedmodule.__init__(self._sh{0}{1})".format("," if len(arglist) else "", argstring))
if self.loadedmodule._init_complete == True:
try:
code_version = self.loadedmodule.version
except:
code_version = None # if module code without version
if self.meta.test_version(code_version):
logger.info("Modules: Loaded module '{}' (class '{}') v{}: {}".format( name, str(self.loadedmodule.__class__.__name__), self.meta.get_version(), self.meta.get_mlstring('description') ) )
self._moduledict[name] = self.loadedmodule
self._modules.append(self._moduledict[name])
return self.loadedmodule
else:
logger.error(f"Module {name} not started: Module version mismatch")
return None
else:
logger.error("Modules: Module '{}' initialization failed, module not loaded".format(classpath.split('.')[1]))
return None
# ------------------------------------------------------------------------------------
# Following (static) methods of the class Modules implement the API for modules in shNG
# ------------------------------------------------------------------------------------
[Doku] @staticmethod
def get_instance():
"""
Returns the instance of the Modules class, to be used to access the modules-api
Use it the following way to access the api:
.. code-block:: python
from lib.module import Modules
modules = Modules.get_instance()
# to access a method (eg. xxx()):
modules.xxx()
:return: modules instance
:rtype: object of None
"""
if _modules_instance == None:
return None
else:
return _modules_instance
[Doku] def return_modules(self):
"""
Returns a list with the names of all loaded modules
:return: list of module names
:rtype: list
"""
l = []
for module_key in self._moduledict.keys():
l.append(module_key)
return l
[Doku] def get_module(self, name):
"""
Returns the module object for the module named by the parameter
or None, if the named module is not loaded
:param name: Name of the module to return
:type name: str
:return: list of module names
:rtype: object
"""
return self._moduledict.get(name)
[Doku] def start(self):
"""
Start all modules
Call start routine of module in case the module wants to start any threads
"""
logger.info('Start Modules')
for module in self.return_modules():
logger.debug('Starting {} Module'.format(module))
self.m = self.get_module(module)
self.m.start()
[Doku] def stop(self):
"""
Stop all modules
Call stop routine of module to clean up in case the module has started any threads
"""
logger.info('Stop Modules')
module_list = self.return_modules()
# stop modules in revered order (module started first is stopped last)
module_list.reverse()
for module in module_list:
logger.debug('Stopping {} Module'.format(module))
self.m = self.get_module(module)
try:
self.m.stop()
# except:
# pass
except Exception as e:
logger.warning("Error while stopping module '{}'\n-> {}".format(module, e))