/usr/lib/python2.7/dist-packages/pbcommand/cli/core.py is in python-pbcommand 0.2.17-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 | """ New Commandline interface that supports ResolvedToolContracts and emitting ToolContracts
There's three use cases.
- running from an argparse instance
- running from a Resolved Tool Contract (RTC)
- emitting a ToolContract (TC)
Going to do this in a new steps.
- de-serializing of RTC (I believe this should be done via avro, not a new random JSON file. With avro, the java, c++, classes can be generated. Python can load the RTC via a structure dict that has a well defined schema)
- get loading and running of RTC from commandline to call main func in a report.
- generate/emit TC from a a common commandline parser interface that builds the TC and the standard argparse instance
"""
import argparse
import json
import logging
import time
import traceback
import sys
from pbcommand.common_options import (RESOLVED_TOOL_CONTRACT_OPTION,
EMIT_TOOL_CONTRACT_OPTION)
from pbcommand.models import PbParser
from pbcommand.pb_io.tool_contract_io import load_resolved_tool_contract_from
def get_default_argparser(version, description):
"""
Everyone MUST use this to create an instance on a argparser python parser.
:param version:
:param description:
:return:
:rtype: ArgumentParser
"""
p = argparse.ArgumentParser(version=version,
description=description,
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
return p
def _pacbio_main_runner(alog, setup_log_func, func, *args, **kwargs):
"""
Runs a general func and logs results. The return type is expected to be an (int) return code.
:param alog: a log instance
:param func: a cli exe func, must return an int exit code. func(args) => Int, where args is parsed from p.parse_args()
:param args: parsed args from parser
:return: Exit code of callable func
:rtype: int
"""
started_at = time.time()
pargs = args[0]
level = logging.INFO
# Assuming that the log_level might not be an added option.
if hasattr(pargs, 'log_level'):
level = logging.getLevelName(pargs.log_level)
log_options = dict(level=level)
# The Setup log func must adhere to the pbcommand.utils.setup_log func
# signature
setup_log_func(alog, **log_options)
try:
# the code in func should catch any exceptions. The try/catch
# here is a fail safe to make sure the program doesn't fail
# and the makes sure the exit code is logged.
return_code = func(*args, **kwargs)
run_time = time.time() - started_at
except Exception as e:
run_time = time.time() - started_at
alog.error(e, exc_info=True)
traceback.print_exc(sys.stderr)
# We should have a standard map of exit codes to Int
if isinstance(e, IOError):
return_code = 1
else:
return_code = 2
_d = dict(r=return_code, s=run_time)
alog.info("exiting with return code {r} in {s:.2f} sec.".format(**_d))
return return_code
def _get_resolved_tool_contract_from_argv(argv):
"""
Extract the resolved tool contract path from the raw argv
There are two cases
--resolved-tool-contract-path=/path/to/tool_contract.json
--resolved-tool-contract-path /path/to/tool_contract.json
:param argv:
:rtype: str
:raises: ValueError
:return: Path to Manifest
"""
# this is a lackluster implementation. FIXME.
m_str = RESOLVED_TOOL_CONTRACT_OPTION
error = ValueError("Unable to extract resolved tool contract from commandline args {a}. Expecting {m}=/path/to/file.json".format(a=argv, m=m_str))
tool_contract_path = None
nargv = len(argv)
# Provided the --resolved-tool-contract /path/to/tool_contract_path.json
if m_str in argv:
for i, a in enumerate(argv):
# print i, nargv, a
if a.startswith(m_str):
if (i + 1) <= nargv:
tool_contract_path = argv[i + 1]
break
else:
raise error
# Provided the --resolved-tool-contract=/path/to/tool_contract_path.json
m_str_eq = m_str + "="
for i in argv:
if i.startswith(m_str_eq):
tool_contract_path = i.split(m_str_eq)[-1]
break
if tool_contract_path is None:
raise error
return tool_contract_path
def pacbio_args_runner(argv, parser, args_runner_func, alog, setup_log_func):
# For tools that haven't yet implemented the ToolContract API
args = parser.parse_args(argv)
return _pacbio_main_runner(alog, setup_log_func, args_runner_func, args)
def pacbio_args_or_contract_runner(argv,
parser,
args_runner_func,
contract_tool_runner_func,
alog, setup_log_func):
"""
For tools that understand resolved_tool_contracts, but can't emit
tool contracts (they may have been written by hand)
:param parser: argparse Parser
:type parser: ArgumentParser
:param args_runner_func: func(args) => int signature
:param contract_tool_runner_func: func(tool_contract_instance) should be
the signature
:param alog: a python log instance
:param setup_log_func: func(log_instance) => void signature
:return: int return code
:rtype: int
"""
# circumvent the argparse parsing by inspecting the raw argv, then manually
# parse out the resolved_tool_contract path. Not awesome, but the only way to skip the
# parser.parse_args(args) machinery
if any(a.startswith(RESOLVED_TOOL_CONTRACT_OPTION) for a in argv):
print "Attempting to Load resolved tool contract from {a}".format(a=argv)
# FIXME need to catch the exception if raised here before the _pacbio_main_runner is called
resolved_tool_contract_path = _get_resolved_tool_contract_from_argv(argv)
resolved_tool_contract = load_resolved_tool_contract_from(resolved_tool_contract_path)
r = _pacbio_main_runner(alog, setup_log_func, contract_tool_runner_func, resolved_tool_contract)
# alog.info("Completed running resolved contract. {c}".format(c=resolved_tool_contract))
return r
else:
# tool was called with the standard commandline invocation
return pacbio_args_runner(argv, parser, args_runner_func, alog,
setup_log_func)
def pbparser_runner(argv,
parser,
args_runner_func,
contract_runner_func,
alog,
setup_log_func):
"""Run a Contract or emit a contract to stdout."""
if not isinstance(parser, PbParser):
raise TypeError("Only supports PbParser.")
arg_parser = parser.arg_parser.parser
# extract the contract
tool_contract = parser.to_contract()
if EMIT_TOOL_CONTRACT_OPTION in argv:
# print tool_contract
x = json.dumps(tool_contract.to_dict(), indent=4)
print x
else:
return pacbio_args_or_contract_runner(argv, arg_parser, args_runner_func, contract_runner_func, alog, setup_log_func)
|