feat: add python cli support

workflow/__init__.py

@@ -0,0 +1 @@
+__version__ = "1.2.0"

workflow/cli_utils.py

@@ -0,0 +1,116 @@
+import sys
+import os
+import subprocess
+import click
+import logging
+
+# Get the directory path of this file
+base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+workflow_dir = os.path.join(base_dir, "workflow")
+
+logging.basicConfig(level=logging.INFO)
+
+def get_snakefile(file="Snakefile"):
+    snake_file = os.path.join(workflow_dir, file)
+    if not os.path.exists(snake_file):
+        sys.exit("Unable to locate the  Snakefile;  tried %s" % snake_file)
+    return snake_file
+
+def get_configfile(file="config.yaml"):
+    config_file = os.path.join(workflow_dir, "config", file)
+    if not os.path.exists(config_file):
+        sys.exit("Unable to locate the config.yaml file;  tried %s" % config_file)
+    return config_file
+
+def show_help_message():
+    message = (
+        "\nUsage examples:\n"
+        "\n    kgwasflow run [OPTIONS]         Run the kGWASflow workflow\n"
+        "\n    kgwasflow test [OPTIONS]        Run the kGWASflow test\n"
+        "\n    kgwasflow --help"
+        "\n\nRun examples:"
+        "\n\n1. Run kGWASflow with the default config file (../config/config.yaml), default snakemake arguments and 16 threads:\n"
+        "\n    kgwasflow run -t 16 --snake-default"
+        "\n\n2. Run kGWASflow with a custom config file (path/to/custom_config.yaml) and default settings:\n"
+        "\n    kgwasflow run -t 16 -c path/to/custom_config.yaml"
+        "\n\n3. Run kGWASflow with user defined output directory:\n"
+        "\n    kgwasflow run -t 16 --output path/to/output_dir"
+        "\n\n4. Run kGWASflow in dryrun mode to see what tasks would be executed without actually running them:\n"
+        "\n    kgwasflow run -t 16 -n"
+        "\n\n5. Run kGWASflow using mamba as the conda frontend:\n"
+        "\n    kgwasflow run -t 16 --conda-frontend mamba"
+        "\n\n6. Run kGWASflow and generate an HTML report:\n"
+        "\n    kgwasflow run -t 16 --generate-report"
+        "\n\nTest examples:"
+        "\n\n1. Run the kGWASflow test in dryrun mode to see what tasks would be executed:\n"
+        "\n    kgwasflow test -t 16 -n"
+        "\n\n2. Run the kGWASflow test using the test config file with 16 threads:\n"
+        "\n    kgwasflow test -t 16"
+        "\n\n3. Run the kGWASflow test and define the test output folder:\n"
+        "\n    kgwasflow test -t 16  --output path/to/test_output_dir"
+    )
+    return message
+
+
+def show_ascii_art():
+    click.echo("""
+    \b           
+     _     _______          __      _____  __ _               
+    | |   / ____\ \        / /\    / ____|/ _| |              
+    | | _| |  __ \ \  /\  / /  \  | (___ | |_| | _____      __
+    | |/ / | |_ | \ \/  \/ / /\ \  \___ \|  _| |/ _ \ \ /\ / /
+    |   <| |__| |  \  /\  / ____ \ ____) | | | | (_) \ V  V / 
+    |_|\_\\_____|   \/  \/_/    \_\_____/|_| |_|\___/ \_/\_/  
+    \b
+    kGWASflow: A Snakemake Workflow for k-mers Based GWAS
+    """)
+
+def run_snake(snakefile, config_file, threads, output, conda_frontend, dryrun, generate_report, snake_default, rerun_triggers, verbose, unlock, snakemake_args):
+    # Define the command to run snakemake
+    cmd = ['snakemake', '--use-conda', '--conda-frontend', conda_frontend, '--cores', str(threads)]
+
+    if snakefile:
+        cmd += ['--snakefile', snakefile]
+
+    # if config file is provided, use it
+    if config_file:
+        cmd += ['--configfile', config_file]
+
+    # if output directory is provided, use it
+    if output:
+        if not os.path.exists(output):
+            os.makedirs(output)
+        cmd += ['--directory', output]
+
+    if dryrun:
+        cmd.append('--dryrun')
+
+    if generate_report:
+        if dryrun:
+            cmd.append('--report')
+            cmd.append('kGWASflow-report.html')
+        if not dryrun:
+            cmd.append('--dryrun')
+            cmd.append('--report')
+            cmd.append('kGWASflow-report.html')
+
+    if snakemake_args:
+        cmd += snakemake_args
+
+    if rerun_triggers:
+        cmd += ['--rerun-triggers'] + list(rerun_triggers)
+
+    if snake_default:
+        default_snakemake_args = ["--rerun-incomplete", "--printshellcmds", "--nolock", "--show-failed-logs"]
+        cmd += default_snakemake_args
+
+    if verbose:
+        cmd.append('--verbose')
+
+    if unlock:
+        cmd.append('--unlock')
+
+    try:
+        subprocess.run(cmd, check=True)
+    except subprocess.CalledProcessError as e:
+        logging.error("Error running Snakemake: {}".format(e))

workflow/kgwasflow.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python3
+
+import click
+import os
+
+from workflow import __version__
+from .cli_utils import run_snake, get_snakefile, get_configfile, show_ascii_art, show_help_message, workflow_dir, base_dir
+
+# kgwasflow_dir = os.path.join(os.getcwd())
+
+@click.group(context_settings=dict(help_option_names=["-h", "--help"]))
+@click.version_option(__version__)
+def cli():
+    """kGWASflow is a Snakemake workflow for performing k-mers-based GWAS.
+    \b
+    For more options, run:
+    kgwasflow --help
+    kgwasflow run --help
+    or,
+    kgwasflow test --help"""
+    pass
+
+def common_options(func):
+    """Decorator for common options."""
+    options = [
+        click.option('-s', '--snakefile', type=click.Path(dir_okay=True, writable=True, resolve_path=True), help='Path to the Snakefile.'),
+        click.option('-c', '--config-file', type=click.Path(dir_okay=True, writable=True, resolve_path=True), help='Path to the config.yaml file'),
+        click.option('-t', '--threads', default=8, type=int, help='Number of threads (default: 8).', show_default=True),
+        click.option('-o', '--output', help="Output directory.", type=click.Path(dir_okay=True, writable=True, readable=True)),
+        click.option('--conda-frontend', default='conda', type=str, help='Conda frontend to use.'),
+        click.option('-n', '--dryrun', is_flag=True, default=False, show_default=True, help='Dry run. Do not execute the workflow, but show which jobs would be executed.'),
+        click.option('-r', '--generate-report', is_flag=True, default=False, help='Create a kGWASflow HTML report.', show_default=True),
+        click.option('--snake-default', is_flag=True, default=False, help='Useful default snakemake arguments.', show_default=True),
+        click.option('--rerun-triggers', multiple=True, default= ["mtime", "params", "input", "software-env", "code"], help='Rerun all jobs that have at least one of the specified trigger files changed.', show_default=True),
+        click.option('--unlock', is_flag=True, help='Unlock the workflow if it is locked.'),
+        click.option('-v', '--verbose', is_flag=True, default=False, help='Increase output verbosity.'),
+        click.argument("snakemake_args", nargs=-1, type=click.UNPROCESSED),
+    ]
+    for option in reversed(options):
+        func = option(func)
+    return func
+
+@click.command(epilog=show_help_message(), context_settings=dict(help_option_names=["-h", "--help"], ignore_unknown_options=True))
+@common_options
+def run(snakefile, config_file, **kwargs):
+    """Run kGWASflow workflow."""
+    if not snakefile:
+        snakefile = get_snakefile()
+    if not config_file:
+        config_file = get_configfile()
+    run_snake(snakefile, config_file, **kwargs)
+
+@common_options
+@click.command(epilog=show_help_message(), context_settings=dict(help_option_names=["-h", "--help"], ignore_unknown_options=True))
+def test(snakefile, config_file, **kwargs):
+    """Test kGWASflow workflow."""
+    if not snakefile:
+        snakefile = get_snakefile()
+
+    test_config_file = os.path.join(workflow_dir, "test", "config_ecoli", "config.yaml")
+    if not config_file:
+        config_file = test_config_file
+    run_snake(snakefile, config_file, **kwargs)
+
+cli.add_command(run)
+cli.add_command(test)
+
+def main():
+    show_ascii_art()
+    cli()
+
+if __name__ == "__main__":
+    main()