From ae68681481f9ed43b0672809ecf83b2de60bb877 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Dausseur?= Date: Fri, 28 Apr 2023 09:48:12 +0200 Subject: [PATCH] parameters management improved --- src/pyappengine/__init__.py | 268 +++++++++++------------------------- 1 file changed, 83 insertions(+), 185 deletions(-) diff --git a/src/pyappengine/__init__.py b/src/pyappengine/__init__.py index d912527..8474c02 100644 --- a/src/pyappengine/__init__.py +++ b/src/pyappengine/__init__.py @@ -11,6 +11,7 @@ from threading import Thread, Lock import inspect from threading import Thread + def is_number(s): try: float(s) @@ -19,10 +20,12 @@ def is_number(s): pass return False + class Commands(Thread): defmod = None + precmd = "cmd_" - def __init__(self, config, log): + def __init__(self, config: ConfigParser, log: logging.Handler) -> None: super().__init__() self.threaded = False self.cmods = {} @@ -31,133 +34,115 @@ class Commands(Thread): self.stopped = False self.nickname = None if not (self.config is None): - self.nickname = self.config.get('nickname') + self.nickname = self.config.get("alias") def val_to_print(self, val, dig=1): if isinstance(val, str): return val if isinstance(val, float): - f = '{:.' + str(int(dig)) + 'f}' + f = "{:." + str(int(dig)) + "f}" return f.format(val) else: return str(val) - def info(self, msg): if self.log: self.log.info(msg) else: - print('info: '+msg) + print("info: " + msg) def debug(self, msg): if self.log: self.log.debug(msg) else: - print('debug: '+msg) + print("debug: " + msg) def warning(self, msg): if self.log: self.log.warning(msg) else: - print('warning: '+msg) + print("warning: " + msg) def error(self, msg): if self.log: self.log.error(msg) else: - print('error: '+msg) + print("error: " + msg) def stop(self): self.stopped = True def list_modules(self): success = True - ret = '' + ret = "" for module in self.cmods.keys(): - ret = ret + module + '\n' + ret = ret + module + "\n" return success, ret.strip() - def _execute_command(self, cmd): - c = cmd.split(',', maxsplit=1) - key = "cmd_" + c[0].strip().lower().replace(' ', '_') - params = [] - - #parameters - if len(c) > 1: - for p in c[1].split(','): - p = p.strip() - if is_number(p): - params.append(float(p)) - else: - params.append(p) - params = tuple(params) - + def _execute_command(self, method: str, *args, **kwargs) -> tuple: success = False ret = "function not found" - if hasattr(self, key) and inspect.ismethod(getattr(self, key)): + if hasattr(self, self.precmd + method) and inspect.ismethod( + getattr(self, self.precmd + method) + ): self.lock.acquire() try: - f = getattr(self, key) - ret = f(*params) + f = getattr(self, self.precmd + method) + ret = f(*args, **kwargs) success = True - self.log.info('function "{}" of module "{}" was called'.format(key, self.name)) + self.log.info( + 'function "{}" of module "{}" was called'.format(method, self.name) + ) except TypeError: - self.log.error('function "{}" of module "{}" wrong parameters'.format(key, self.name)) - ret = 'param error' + self.log.error( + 'function "{}" of module "{}" wrong parameters'.format( + method, self.name + ) + ) + ret = "param error" except: - self.log.error('function "{}" of module "{}" crashed'.format(key, self.name)) - ret = 'function crash' + self.log.error( + 'function "{}" of module "{}" crashed'.format(method, self.name) + ) + ret = "function crash" self.lock.release() if ret is None: - ret = '' + ret = "" return success, ret.strip() - def execute_command(self, cmd): + def execute_command(self, module: str, method: str, *args, **kwargs): # isolate the module called success = False - c = cmd.split('.', maxsplit=1) - if ((len(c) == 1) and (self.defmod is None)) or (len(c) < 1): - ret = 'Wrong call syntax. It should be: ".=param,param2..."' - self.log.error(ret) - return success, ret - - if (len(c) == 1) and (c[0].lower() == 'help'): - return self.list_modules() - - elif len(c) == 1: - module = self.defmod - command = c[0] - else: - module = c[0].lower() - command = c[1] + if module == "": + module = self.config["default"] if not (module in self.cmods.keys()): - m = module.replace('_','-') + m = module.replace("_", "-") if not (m in self.cmods.keys()): ret = 'module "{}" not found'.format(module) self.log.error(ret) return success, ret - return self.cmods[module]._execute_command(command) + return self.cmods[module]._execute_command(method, *args, **kwargs) - def cmd_help(self,*args): - '''Help of module commands. + def cmd_help(self, *args, **kwargs): + """Help of module commands. Params: - if No param: list of commands - - otherwise: help of the first arg''' + - otherwise: help of the first arg""" - ret = '' + ret = "" if len(args) == 0: cmds = inspect.getmembers(self, predicate=inspect.ismethod) - cmds = [x for x in cmds if x[0].startswith('cmd_')] + cmds = [x for x in cmds if x[0].startswith("cmd_")] for m in cmds: - ret = ret + m[0][len('cmd_'):] + '\n' + ret = ret + m[0][len("cmd_") :] + "\n" else: if isinstance(args[0], str): - cmd = 'cmd_'+args[0] + cmd = "cmd_" + args[0] c = getattr(self, cmd, None) if c: r = inspect.getdoc(c) @@ -166,137 +151,38 @@ class Commands(Thread): else: ret = 'No documentation for "' + args[0] + '"' - if ret == '': - ret = 'No command with this name...' + if ret == "": + ret = "No command with this name..." return ret - def error(self, msg): - if self.log: - self.log.error(msg) - else: - print('error: '+msg) - - def stop(self): - self.stopped = True - - def list_modules(self): - success = True - ret = '' - for module in self.cmods.keys(): - ret = ret + module + '\n' - return success, ret.strip() - - def _execute_command(self, cmd): - c = cmd.split(',', maxsplit=1) - key = "cmd_" + c[0].strip().lower().replace(' ', '_') - params = [] - - #parameters - if len(c) > 1: - for p in c[1].split(','): - p = p.strip() - if is_number(p): - params.append(float(p)) - else: - params.append(p) - params = tuple(params) - - success = False - ret = "function not found" - if hasattr(self, key) and inspect.ismethod(getattr(self, key)): - self.lock.acquire() - try: - f = getattr(self, key) - ret = f(*params) - success = True - self.log.info('function "{}" of module "{}" was called'.format(key, self.name)) - except TypeError: - self.log.error('function "{}" of module "{}" wrong parameters'.format(key, self.name)) - ret = 'param error' - except: - self.log.error('function "{}" of module "{}" crashed'.format(key, self.name)) - ret = 'function crash' - self.lock.release() - - if ret is None: - ret = '' - return success, ret.strip() - - def execute_command(self, cmd): - # isolate the module called - success = False - c = cmd.split('.', maxsplit=1) - if ((len(c) == 1) and (self.defmod is None)) or (len(c) < 1): - ret = 'Wrong call syntax. It should be: ".=param,param2..."' - - self.log.error(ret) - return success, ret - - if (len(c) == 1) and (c[0].lower() == 'help'): - return self.list_modules() - - elif len(c) == 1: - module = self.defmod - command = c[0] - else: - module = c[0].lower() - command = c[1] - - if not (module in self.cmods.keys()): - m = module.replace('_','-') - if not (m in self.cmods.keys()): - ret = 'module "{}" not found'.format(module) - self.log.error(ret) - return success, ret - - return self.cmods[module]._execute_command(command) - - def cmd_help(self,*args): - '''Help of module commands. - Params: - - if No param: list of commands - - otherwise: help of the first arg''' - - ret = '' - - if len(args) == 0: - cmds = inspect.getmembers(self, predicate=inspect.ismethod) - cmds = [x for x in cmds if x[0].startswith('cmd_')] - for m in cmds: - ret = ret + m[0][len('cmd_'):] + '\n' - else: - if isinstance(args[0], str): - cmd = 'cmd_'+args[0] - c = getattr(self, cmd, None) - if c: - r = inspect.getdoc(c) - if r: - ret = r - else: - ret = 'No documentation for "' + args[0] + '"' - - if ret == '': - ret = 'No command with this name...' - - return ret class CommandsLoader: def __init__(self, config: ConfigParser, log: logging.Handler, modpath: str): self.config = config self.modpath = modpath - Commands.defmod = self.config['general'].get('default') + Commands.defmod = self.config["general"].get("default") + precmd = self.config["general"].get("methods_prefix") + if not precmd is None: + Commands.precmd = precmd + self.precmds = "cmds_" + precmds = self.config["general"].get("modules_prefix") + if not precmds is None: + self.precmds = precmds self.log = log self.lock = Lock() self._load_commands() self._set_cmods() - self._load_dependencies() def _load_commands(self): cmds = {} with os.scandir(self.modpath) as it: for entry in it: - if entry.name.startswith('cmds_') and entry.name.endswith('.py') and entry.is_file(): + if ( + entry.name.startswith(self.precmds) + and entry.name.endswith(".py") + and entry.is_file() + ): obj = self._load_module(entry.name[:-3]) if not obj is None: nmod = obj.nickname @@ -306,11 +192,14 @@ class CommandsLoader: self.log.info('module "{}" loaded'.format(nmod)) else: self.log.error( - 'The module "{}" could not be loaded'.format(entry.name[:-3])) + 'The module "{}" could not be loaded'.format( + entry.name[:-3] + ) + ) self.cmods = cmds - def _load_module(self, name): - module = import_module("."+name, 'commands') + def _load_module(self, name: str): + module = import_module(name) members = inspect.getmembers(module, inspect.isclass) conf = None if name[5:] in self.config.sections(): @@ -318,7 +207,7 @@ class CommandsLoader: obj = None for n, c in members: - if issubclass(c, Commands) and (n != 'Commands'): + if issubclass(c, Commands) and (n != "Commands"): obj = c(conf, self.log) obj.log = self.log obj.lock = self.lock @@ -338,9 +227,13 @@ class CommandsLoader: setattr(v, p, self.cmods[p]) else: self.log.error( - 'Dependency "{}" of module "{}" could not be satisfied'.format(p, k)) + 'Dependency "{}" of module "{}" could not be satisfied'.format( + p, k + ) + ) def start(self): + self._load_dependencies() for k, v in self.cmods.items(): if v.threaded: v.start() @@ -352,10 +245,16 @@ class CommandsLoader: class AppEngine: - def __init__(self, app_name: str, conf_file : str="", log_file: str="", debug: bool=False) -> None: + def __init__( + self, + app_name: str, + conf_file: str = "", + log_file: str = "", + debug: bool = False, + ) -> None: self.app_name = app_name self.conf = ConfigParser() - if conf_file != '': + if conf_file != "": self.parse_config(conf_file) self.def_log(log_file) @@ -368,11 +267,11 @@ class AppEngine: if os.path.exists(cf) and os.path.isfile(cf): self.conf.read(cf) else: - raise Exception('Configuration file not found') + raise Exception("Configuration file not found") - def def_log(self, log_file): + def def_log(self, log_file: str): self.log = logging.getLogger(self.app_name) - if log_file != '': + if log_file != "": fname = log_file is_writeable = False @@ -391,8 +290,7 @@ class AppEngine: else: self.log.addHandler(journal.JournalHandler()) - - def exec(self, modpath: str=""): + def exec(self, modpath: str = ""): cl = CommandsLoader(self.conf, self.log, modpath) cl.start() cl.join()