Skip to content

Commit

Permalink
Reimplement click options as argparse arguments
Browse files Browse the repository at this point in the history
  • Loading branch information
sdruskat committed Nov 15, 2023
1 parent 481a507 commit f1e7115
Showing 1 changed file with 41 additions and 150 deletions.
191 changes: 41 additions & 150 deletions src/hermes/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,161 +8,52 @@
"""
This module provides the main entry point for the HERMES command line application.
"""
import logging
import typing as t
import pathlib
from importlib import metadata
import argparse

from hermes import config
from hermes.commands import workflow
from hermes.config import configure, init_logging


def log_header(header, summary=None):
_log = config.getLogger('cli')

dist = metadata.distribution('hermes')
meta = dist.metadata

if header is None:
title = f"{dist.name} workflow ({dist.version})"

_log.info(title)
_log.info("=" * len(title))
_log.info('')

if 'Summary' in meta:
_log.info('%s', meta['Summary'])
_log.info('')

else:
_log.info("%s", header)
if summary:
_log.info("%s", summary)
_log.info('')


class WorkflowCommand(click.Group):
"""
Custom multi-command that implements
- sub-commands in a fixed order,
- automatic call of sub-commands, and
- filtering of commands to be executed.
"""

def __init__(self, *args, **kwargs) -> None:
self._command_order = []
super().__init__(*args, **kwargs)

def add_command(self, cmd: click.Command, name: t.Optional[str] = None) -> None:
"""
Overridden to ensure the order of commands is retained.
"""

cmd_name = name or cmd.name
if cmd_name not in self.commands:
self._command_order.append(cmd_name)
super().add_command(cmd, name)

def list_commands(self, ctx: click.Context) -> t.List[str]:
"""
Overridden to return commands in fixed order.
"""

return self._command_order

def invoke(self, ctx: click.Context) -> t.Any:
"""
Invoke all sub-commands in a given order.
If there is a flag in `ctx.params` with the name of a sub-command that evaluates to `False`,
the command is not invoked.
:param ctx: Context for the command.
"""

# Get the user provided working dir from the --path option or default to current working directory.
working_path = ctx.params.get('path').absolute()

configure(ctx.params.get('config').absolute(), working_path)
init_logging()
log_header(None)

audit_log = logging.getLogger('audit')
audit_log.info("# Running Hermes")
audit_log.info("Running Hermes command line in: %s", working_path)
audit_log.debug("")
audit_log.debug("Invoked `%s` with", ctx.invoked_subcommand or self.name)
audit_log.debug("")
for k, v in ctx.params.items():
audit_log.debug("`--%s`", k)
audit_log.debug(": `%s`", v)
audit_log.debug("")

if ctx.protected_args:
return super().invoke(ctx)

# The following code is largely copied from the base implementation that can be found in
# click.core.MultiCommand.

def _process_result(value: t.Any) -> t.Any:
if self._result_callback is not None:
value = ctx.invoke(self._result_callback, value, **ctx.params)
return value

args = [*ctx.protected_args, *ctx.args]
ctx.args = []
ctx.protected_args = []

with ctx:
contexts = []
for cmd_name in self.list_commands(ctx):
if not ctx.params.get(cmd_name, True):
continue

cmd = self.get_command(ctx, cmd_name)
sub_ctx = cmd.make_context(
cmd_name,
args,
parent=ctx,
allow_extra_args=True,
allow_interspersed_args=False
)
contexts.append(sub_ctx)
args, sub_ctx.args = sub_ctx.args, []

rv = []
for sub_ctx in contexts:
with sub_ctx:
rv.append(sub_ctx.command.invoke(sub_ctx))
return _process_result(rv)


@click.group(cls=WorkflowCommand, invoke_without_command=True)
@click.option(
"--config", default=pathlib.Path('hermes.toml'),
help="Configuration file in TOML format", type=pathlib.Path
)
@click.option("--curate", is_flag=True, default=False)
@click.option("--deposit", is_flag=True, default=False)
@click.option("--postprocess", is_flag=True, default=False)
@click.option("--clean", is_flag=True, default=False)
@click.option('--path', default=pathlib.Path('./'), help='Working path', type=pathlib.Path)
@click.pass_context
def main(ctx: click.Context, *args, **kwargs) -> None:
def main(*args, **kwargs) -> None:
"""
HERMES
This command runs the HERMES workflow or parts of it.
"""

pass


main.add_command(workflow.clean)
main.add_command(workflow.harvest)
main.add_command(workflow.process)
main.add_command(workflow.curate)
main.add_command(workflow.deposit)
main.add_command(workflow.postprocess)
parser = argparse.ArgumentParser(
prog="hermes",
description="This command runs the HERMES workflow or parts of it.",
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
parser.add_argument(
"--config",
default=pathlib.Path("hermes.toml"),
help="Configuration file in TOML format",
type=pathlib.Path,
)
parser.add_argument(
"--curate",
action="store_true",
default=False,
help="Run the 'curate' workflow step",
)
parser.add_argument(
"--deposit",
action="store_true",
default=False,
help="Run the 'deposit' workflow step",
)
parser.add_argument(
"--postprocess",
action="store_true",
default=False,
help="Run the 'postprocess' workflow step",
)
parser.add_argument(
"--clean",
action="store_true",
default=False,
help="Remove cached data",
)
parser.add_argument(
"--path", default=pathlib.Path("./"), help="Working path", type=pathlib.Path
)
args = parser.parse_args()

0 comments on commit f1e7115

Please sign in to comment.