Example CLI using pbsystem and new ToolContract interface

example_cli_app.py

"""Simple CLI dev app for testing"""

import logging
import sys

from pbsystem.common.cmdline.core2 import (get_default_parser,
                                           pacbio_args_or_contract_runner,
                                           add_resolved_tool_contract_option)
from pbsystem.common.utils import setup_log
from pbsystem.common.validators import validate_file
from .tool_contract import ResolvedToolContract

log = logging.getLogger(__name__)

__version__ = '0.1.0'


def get_parser():
    p = get_default_parser(__version__, __doc__)
    p.add_argument("input_txt", type=validate_file, help="Path to TXT file")
    p.add_argument("output_txt", type=str, help="Output TXT fiel to write to")
    p.add_argument('--alpha', type=int, default=6, help="Alpha example option.")
    add_resolved_tool_contract_option(p)
    return p


def run_main(input_file, output_file, alpha):
    """
    Main function entry point to your application (this should be imported
    from your library code)

    :rtype int:
    """
    _d = dict(i=input_file, a=alpha, o=output_file)
    msg = "Running dev_app task. with input:{i} output:{o} and alpha={a}".format(**_d)
    log.info(msg)
    with open(output_file, 'w') as w:
        w.write(msg + "\n")
    return 0


def args_runner(args):
    """Entry point from argparse"""
    return run_main(args.input_txt, args.output_txt, args.alpha)


def resolved_tool_contract_runner(resolved_tool_contract):
    """Run from the resolved contract

    :param resolved_tool_contract:
    :type resolved_tool_contract: ResolvedToolContract
    """

    in_file = resolved_tool_contract.task.input_files[0]
    out_file = resolved_tool_contract.task.output_files[0]
    alpha = 9
    return run_main(in_file, out_file, alpha)


def main(argv=sys.argv):
    # New interface that supports running resolved tool contracts
    log.info("Starting {f} version {v} pbsystem example dev app".format(f=__file__, v=__version__))
    return pacbio_args_or_contract_runner(argv[1:],
                                          get_parser(),
                                          args_runner,
                                          resolved_tool_contract_runner,
                                          log,
                                          setup_log)


if __name__ == '__main__':
    sys.exit(main())