#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import collections
from poppy.core.logger import logger
from poppy.core.generic.signals import Signal
from poppy.core.tools.poppy_argparse import argparse
from poppy.core.generic.metaclasses import ManagerMeta
from poppy.core.pipeline import Pipeline
__all__ = ["Command"]
def after_creation(cls, name):
"""
Some extra things to do after the creation of command classes (and not
instances, this is not an error).
"""
# check that the class has the good attributes
if not hasattr(cls, "__command__"):
logger.error(
"The class {0} has no attribute __command__".format(name)
)
return
if not hasattr(cls, "__parent__"):
logger.error(
"The class {0} has no attribute __parent__".format(name)
)
return
[docs]class CommandManager(object):
"""
A class to manage the available defined commands by the user through the
plugin command classes.
"""
def __init__(self):
"""
Create containers where to store defined commands.
"""
# containers
self.commands = {}
self.instancesByCommand = {}
# tree of dependencies for commands
self.tree = collections.defaultdict(list)
# define the arguments specific to the pipeline
self.pipeline_arguments = {
'settings': (
['--settings'],
{'help': 'The path to the settings file.'},
),
'log_level': (
['-ll', '--log-level'],
{
'help': 'Specifies the amount information that the pipeline should print to the console ',
'choices': ('DEBUG', 'INFO', 'WARNING', 'ERROR')
}
)
}
# signals
self.added = Signal()
self.created = Signal()
self.generation_start = Signal()
self.generation_stop = Signal()
self.run_start = Signal()
self.run_end = Signal()
[docs] def add(self, name, cls):
"""
Adds the command and its class to the manager.
"""
# register
logger.debug("Register the command {0}".format(name))
self.commands[cls.__command__] = cls
# add its dependence in the tree (none parent is the root tree)
self.tree[cls.__parent__].append(cls.__command__)
# send a signal of a new definition
self.added(name, cls)
[docs] def create(self, instance):
"""
Register the instances of the commands by their name.
"""
# register
logger.debug(
"Register command instance {0}".format(instance.__command__)
)
self.instancesByCommand[instance.__command__] = instance
# send the signal of a new instantiation
self.created(instance)
[docs] def generate(self, options):
"""
Given a first base subparser and the parents parser, generate the
parsers for children commands recursively.
"""
# send signal that the generation is starting
self.generation_start(options)
# generate commands
logger.debug("Generating commands hierarchy")
self._generate()
# send signal that the generation stopped
self.generation_stop(options)
def _generate(self, subparser=None, start=None, parsers=None):
# keep trace of the parsers generated by commands
if parsers is None:
self.parsers = {}
# loop over children
for command in self.tree[start]:
# create an instance of the command
instance = self.commands[command]()
# generate the list of parent parsers used by the command if
# provided by the user
if hasattr(instance, "__parent_arguments__"):
parents = [
self.parsers[name]
for name in instance.__parent_arguments__
]
else:
parents = []
# get the parser for the given command
parser = instance.parser(subparser, parents)
# add arguments to the parser from the user
instance.add_arguments(parser)
# by default run the instance when the command is called
parser.set_defaults(run=instance)
# keep trace of the parser
self.parsers[instance.__command__] = parser
# if the command has children, create a subparser and recall the
# method recursively
if instance.has_children():
# generate the subcommands for this command
self._generate(
instance.subparser(parser),
command,
self.parsers,
)
[docs] def launch(self, argv=None):
"""
Launch the commands by parsing the input of the program and then
calling the good command with the good arguments.
"""
# get arguments
args, unknown = self.parse(argv)
if unknown != []:
logger.warning(f'There may be wrong/unused arguments : {unknown}')
# a signal/hook to do something before running the command with the
# parsed arguments
self.run_start(args)
# call the good command
args.run(args)
# a signal/hook to do something after running the command with the
# parsed arguments
self.run_end(args)
[docs] def parse(self, argv=None):
"""
Responsible to parse the command line, and also check the consistency
of the tree of commands for the base command.
"""
# first check that the number of commands linked to the root that does
# nothing is not greater than one. If it is not the case, several
# parsers are defined and thus we are unable to define which one to
# use.
if len(self.tree[None]) > 1:
logger.error(
"Multiple parsers are defined for the commands. Cannot " +
"decide which one to use."
)
raise SystemExit(-1)
elif len(self.tree[None]) <= 0:
logger.error("No commands defined for the root command.")
raise SystemExit(-1)
# else parse the root parser
return self.parsers[self.tree[None][0]].parse_known_args(args=argv)
[docs] def preprocess_args(self, argv):
"""
Preprocess options like --settings and --config.
These options could affect the pipeline behavior, so they must be processed early.
:param argv:
:return: options, args
"""
logger.debug('Preprocessing of settings, log-level, etc. arguments')
parser = argparse.ArgumentParser(description='Preprocess some arguments.', add_help=False, allow_abbrev=False)
for argument in self.pipeline_arguments.values():
parser.add_argument(*argument[0], **argument[1])
return parser.parse_known_args(argv)
def __getitem__(self, name):
"""
Get instance of command by its name.
"""
return self.instancesByCommand[name]
def __contains__(self, name):
"""
Check if the instance of a command is present in the manager.
"""
return name in self.instancesByCommand
class CommandMetaclass(ManagerMeta):
"""
To manage the definition and instantiation of commands.
"""
def __call__(cls, *args, **kwargs):
"""
Make the command instances as singletons.
"""
# check that the instance is already created or not
if cls.__command__ in cls.manager:
return cls.manager[cls.__command__]
# create the instance as usual
return super(CommandMetaclass, cls).__call__(*args, **kwargs)
[docs]class Command(
object,
metaclass=CommandMetaclass,
manager=CommandManager,
after_creation=after_creation,
):
"""
Base class for all accepted commands for the pop command line program.
"""
[docs] def parser(self, subparser, parents):
"""
Return the parser for the command and options that this command must
use. Take as argument the subparser from the parent parser.
"""
return subparser.add_parser(
self.__command_name__,
help=self.__help__,
parents=parents,
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
conflict_handler="resolve",
)
[docs] def add_arguments(self, parser):
"""
Used by the user to add arguments associated to the given parser.
"""
pass
[docs] def subparser(self, parser):
"""
Should return a subparser for this command. Not always called, just
when the command has possibly subcommands to generate.
"""
if hasattr(self, "__command_name__"):
command = self.__command_name__
else:
command = self.__command__
return parser.add_subparsers(
help="Sub-commands available for command {0}".format(command),
)
[docs] def has_children(self):
"""
To know if the command has subcommands.
"""
# check that the command has subcommands
return len(self.manager.tree[self.__command__]) > 0
[docs] def print_help(self):
"""
Print the command help message
"""
Command.manager.parsers[self.__command__].print_help()
def __call__(self, args):
"""
The command is called with its arguments as parameters, in order to
have the possibility to run the succession of commands necessary.
"""
# create a pipeline instance
pipeline = Pipeline(args)
try:
# setup the tasks
self.setup_tasks(pipeline)
# run the pipeline
pipeline.run()
except NotImplementedError:
# if no tasks are defined, print the help message
self.print_help()
[docs] def setup_tasks(self, pipeline):
"""
Set up the task workflow
"""
raise NotImplementedError()
@classmethod
[docs] def add_parent_parser(cls, name, parser):
"""
Used to add a parser with its options and be able to refer from a
command, since the conflict handler of argparse is not well done, as
many other things.
"""
# check that the parser with this name is not already present
if name in cls.manager.parsers:
logger.error(
"Trying to add parser with name {0} while already defined"
)
raise SystemExit(-1)
# add the parser to the collection
cls.manager.parsers[name] = parser
def __repr__(self):
return "Command {0}".format(self.__command__)
def __str__(self):
return self.__command__
# vim: set tw=79 :