parameters management improved
This commit is contained in:
François Dausseur
@@ -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: "<module>.<function>=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: "<module>.<function>=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()
|
||||
|
||||
Reference in New Issue
Block a user