Mixed Parser Example of exposing a subset of your tool interface to tool contract, and adding all options to the raw argparse instance

dev_mixed_app.py

import json
import logging

from pbcommand.models import TaskTypes, FileTypes, get_default_contract_parser

log = logging.getLogger(__name__)

TOOL_ID = "pbcommand.tools.dev_mixed_app"
__version__ = "0.2.0"


def _get_contract_parser():
    """
    Central point of programmatically defining a Parser.
    :rtype: PbParser
    :return: PbParser
    """
    # Number of processors to use
    nproc = 1
    # Log file, tmp dir, tmp file. See ResourceTypes in models
    resource_types = ()
    # Commandline exe to call "{exe}" /path/to/resolved-tool-contract.json
    driver_exe = "python -m pbcommand.cli.example.dev_app --resolved-tool-contract "
    desc = "Dev app for Testing that supports emitting tool contracts"
    task_type = TaskTypes.LOCAL
    p = get_default_contract_parser(TOOL_ID, __version__, desc, driver_exe, task_type, nproc, resource_types)
    return p


def add_rtc_options(p):
    """
    Add all ins/outs and options that will be in both the tool contract and the argparse layer

    :param p:
    :type p: pbcommand.models.PbParser
    :return:
    """
    p.add_input_file_type(FileTypes.CSV, "csv", "Input CSV", "Input csv description")
    p.add_output_file_type(FileTypes.REPORT, "rpt", "Output Report", "Output PacBio Report JSON", "example.report.json")
    p.add_int("pbcommand.task_options.dev_read_length", "read-length", 25, "Length filter", "Min Sequence Length filter")
    return p


def add_argparse_only(p):
    """
    Standard argparse layer

    :param p:
    :type p: argparse.ArgumentParser
    :return:
    """
    p.add_argument("stuff", type=int, default=1234, help="Example option")
    return p


def get_contract_parser():
    p = _get_contract_parser()
    # minimal ins/outs + options exposed at the contract level
    add_rtc_options(p)
    # add all options to the raw argparse instance
    add_argparse_only(p.arg_parser.parser)
    return p


def write_tool_contract(p, output_json_file):
    with open(output_json_file, 'w') as f:
        f.write(json.dumps(p.to_contract()))