Source code for gratools.main

import click
from pathlib import Path
from .Gratools import Gratools
from gratools import description_tools
from .useful_function import validate_percentage_or_int, CustomGroup
from .logger_config import configure_logger

CONTEXT_SETTINGS = dict(ignore_unknown_options=True, max_content_width=800, help_option_names=('-h', '--help'))
LOG_LEVELS = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']

# Définir les options communes
common_options = [
    click.option('--gfa', '-g',
                 type=click.Path(exists=True, file_okay=True, readable=True, resolve_path=True, path_type=Path),
                 required=True, help='GFA file'),
    click.option('--threads', '-t', type=int, default=1, show_default=True, help='Number of threads'),
    click.option('--verbosity', '-vv', default='DEBUG',
                 type=click.Choice(LOG_LEVELS, case_sensitive=False),
                 show_default=True, help='Set the logging verbosity level.'),
    click.option('--log-path', '-l', default='.',
                 type=click.Path(exists=True, file_okay=False, writable=True, resolve_path=True, path_type=Path),
                 show_default=True, help='Path to the log files.')
]




def apply_common_options(func):
    for option in reversed(common_options):
        func = option(func)
    return func



# Décorateur pour appliquer la configuration du logger


def configure_logging(func):
    def wrapper(ctx, *args, **kwargs):
        verbosity = ctx.params.get('verbosity', 'INFO').upper()
        log_path = ctx.params.get('log_path', Path(".").resolve())
        threads = ctx.params.get('threads', 1)
        logger = configure_logger('GraTools', log_path, verbosity)
        logger.info(description_tools)
        logger.info(f'Verbosity: {verbosity}')
        logger.info(f'Number of threads: {threads}')
        logger.info(f'Log file path: {log_path}')
        ctx.obj = {'logger': logger}
        return func(ctx, *args, **kwargs)

    return wrapper



@click.group(name="gratools", cls=CustomGroup, help="", context_settings=CONTEXT_SETTINGS,
             invoke_without_command=True, no_args_is_help=True)
@click.version_option("1.0.0", "-v", "--version", message="%(prog)s, version %(version)s")
@click.pass_context
def main_command(ctx):
    pass
    # if ctx.invoked_subcommand is None:
    #     click.secho(description_tools, fg='cyan')
    #     click.echo(ctx.get_help())


@main_command.command("show_samples", short_help="Show sample names on GFA files", context_settings=CONTEXT_SETTINGS,
                      no_args_is_help=True)
@apply_common_options
@click.pass_context
@configure_logging
def show_samples(ctx, gfa, verbosity, threads, log_path):
    Gratools(gfa_path=gfa, threads=threads,
             logger=ctx.obj['logger']).display_sample_names()


@main_command.command("show_chr", cls=CustomGroup, short_help="Show sample names and chromosomes on GFA files",
                      context_settings=CONTEXT_SETTINGS, no_args_is_help=True)
@apply_common_options
@click.option('--full/--no-full', is_flag=True, default=False, show_default=True,
              help='Show all fragments of chromosomes')
@click.pass_context
@configure_logging
def show_chr(ctx, gfa, verbosity, threads, log_path, full):
    gratools = Gratools(gfa_path=gfa, threads=threads, logger=ctx.obj['logger'])
    if full:
        gratools.display_full_chromosome_data()
    else:
        gratools.display_chromosome_names()


@main_command.command("extract_sub_graph", cls=CustomGroup, short_help="Extract sub GFA files", context_settings=CONTEXT_SETTINGS,
                      no_args_is_help=True)
@apply_common_options
@click.option('--sample', '-sn', type=str, required=True, help='Sample')
@click.option('--chr', '-c', type=str, required=True, help='Chromosome')
@click.option('--start', '-s', type=int, default=0, show_default=True, help='Start')
@click.option('--end', '-e', type=int, help='End')
@click.option('--samples_list', '-l',
              type=click.Path(exists=True, file_okay=True, readable=True, resolve_path=True, path_type=Path),
              help='Path to file with samples list (one per line)')
@click.option('--build_fasta/--no-build_fasta', is_flag=True, default=False, show_default=True,
              help='Build fasta file of sub graph')
@click.pass_context
@configure_logging
def extract_sub_graph(ctx, gfa, verbosity, threads, log_path, sample, chr, start, end, samples_list, build_fasta):
    gratools = Gratools(gfa_path=gfa, threads=threads, logger=ctx.obj['logger'])
    gratools.extract_sub_graph(sample_name=sample, chromosome=chr, start=start, stop=end,
                               samples_list_path=samples_list, build_fasta=build_fasta)
    gratools.concat_generate_sub_graph()
    if build_fasta:
        gratools.generate_fasta()


@main_command.command("get_fasta", cls=CustomGroup, short_help="Extract fasta from graph files", context_settings=CONTEXT_SETTINGS,
                      no_args_is_help=True)
@apply_common_options
@click.option('--sample', '-sn', type=str, required=True, help='Sample')
@click.option('--chr', '-c', type=str, required=True, help='Chromosome')
@click.option('--start', '-s', type=int, default=0, show_default=True, help='Start')
@click.option('--end', '-e', type=int, help='End')
@click.option('--samples_list', '-l',
              type=click.Path(exists=True, file_okay=True, readable=True, resolve_path=True, path_type=Path),
              help='Path to file with samples list (one per line)')
@click.pass_context
@configure_logging
def get_fasta(ctx, gfa, verbosity, threads, log_path, sample, chr, start, end, samples_list):
    gratools = Gratools(gfa_path=gfa, threads=threads, logger=ctx.obj['logger'])
    gratools.extract_sub_graph(sample_name=sample, chromosome=chr, start=start, stop=end,
                               samples_list_path=samples_list, build_fasta=True)
    gratools.generate_fasta()


@main_command.command("segments_info", cls=CustomGroup, short_help="Shared and specific nodes", context_settings=CONTEXT_SETTINGS,
                      no_args_is_help=True)
@apply_common_options
@click.option('--shared_samples_min', '-sh', callback=validate_percentage_or_int, default=None, required=True,
              show_default=True, help='Minimum samples or percentage')
@click.option('--specific_samples_max', '-sp', callback=validate_percentage_or_int, default=None, required=True,
              show_default=True, help='Maximum samples or percentage')
@click.option('--filter_len', '-f', type=int, default=0, show_default=True, help='Minimum segments length')
@click.pass_context
@configure_logging
def segments_info(ctx, gfa, verbosity, threads, log_path, shared_samples_min, specific_samples_max, filter_len):
    gratools = Gratools(gfa_path=gfa, threads=threads, logger=ctx.obj['logger'])
    gratools.segments_info(shared_min=shared_samples_min, specific_max=specific_samples_max,
                           filter_len=filter_len)


@main_command.command("specific_and_shared_segments", cls=CustomGroup, short_help="Shared and specific nodes",
                      context_settings=CONTEXT_SETTINGS, no_args_is_help=True)
@apply_common_options
@click.option('--sample_list_a_path', '-s',
              type=click.Path(exists=True, file_okay=True, readable=True, resolve_path=True, path_type=Path),
              required=True, help='Minimum samples or percentage')
@click.option('--sample_list_b_path', '-sp',
              type=click.Path(exists=True, file_okay=True, readable=True, resolve_path=True, path_type=Path),
              help='Maximum samples or percentage')
@click.option('--filter_len', '-f', type=int, default=0, show_default=True, help='Minimum segments length')
@click.pass_context
@configure_logging
def specific_and_shared_segments(ctx, gfa, verbosity, threads, log_path, sample_list_a_path, sample_list_b_path,
                                 filter_len):
    gratools = Gratools(gfa_path=gfa, threads=ctx.obj['threads'], logger=ctx.obj['logger'])
    gratools.specific_and_shared_segments(sample_list_a_path=sample_list_a_path,
                                          sample_list_b_path=sample_list_b_path,
                                          filter_len=filter_len)


if __name__ == "__main__":
    main_command()