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
|
"""
Additional utilities for running command-line apps - most of these do not apply
to tool-contract-driven programs. (Ported from pbsmrtpipe)
"""
import argparse
import logging
import os
import platform
import time
import traceback
from pbcommand.validators import validate_file, validate_fofn
from pbcommand.utils import setup_log
log = logging.getLogger(__name__)
def subparser_builder(subparser, subparser_id,
description, options_func, exe_func):
"""
Util to add subparser options
:param subparser:
:param subparser_id:
:param description:
:param options_func: Function that will add args and options to Parser instance F(subparser) -> None
:param exe_func: Function to run F(args) -> Int
:return:
"""
p = subparser.add_parser(subparser_id, help=description)
options_func(p)
p.set_defaults(func=exe_func)
return p
def add_debug_option(p):
p.add_argument('--debug', action='store_true',
help="Send logging info to stdout.")
return p
def _validate_output_dir_or_get_default(value):
if value is None:
return os.getcwd()
else:
if os.path.exists(value):
return os.path.abspath(value)
else:
os.mkdir(value)
return os.path.abspath(value)
def add_output_dir_option(p):
p.add_argument(
'-o',
'--output-dir',
type=_validate_output_dir_or_get_default,
default=os.getcwd(),
help="Output directory.")
return p
def _add_input_file(args_label, type_, help_):
def _wrapper(p):
p.add_argument(args_label, type=type_, help=help_)
return p
return _wrapper
add_fasta_output = _add_input_file(
"fasta_out", str, "Path to output Fasta File")
add_fasta_input = _add_input_file(
"fasta_in",
validate_file,
"Path to Input FASTA File")
add_fastq_output = _add_input_file(
"fastq_out", str, "Path to output Fastq File")
add_fastq_input = _add_input_file(
"fastq_in",
validate_file,
"Path to Input FASTQ File")
add_fofn_input = _add_input_file(
"fofn_in",
validate_fofn,
"Path to Input FOFN (File of file names) File")
add_fofn_output = _add_input_file("fofn_out", str, "Path to output FOFN.")
add_report_output = _add_input_file(
"json_report", str, "Path to PacBio JSON Report")
add_subread_input = _add_input_file(
"subread_ds",
validate_file,
"Path to PacBio Subread DataSet XML")
add_ds_reference_input = _add_input_file(
"reference_ds",
validate_file,
"Path to PacBio Subread DataSet XML")
def args_executer(args):
"""
:rtype int
"""
try:
return_code = args.func(args)
except Exception as e:
log.error(e, exc_info=True)
import sys
traceback.print_exc(sys.stderr)
if isinstance(e, IOError):
return_code = 1
else:
return_code = 2
return return_code
def main_runner(argv, parser, exe_runner_func, setup_log_func, alog):
"""
Fundamental interface to commandline applications
"""
started_at = time.time()
args = parser.parse_args(argv)
# log.debug(args)
# setup log
_have_log_setup = False
if hasattr(args, 'quiet') and args.quiet:
setup_log_func(alog, level=logging.ERROR)
elif getattr(args, 'verbosity', None) is not None and args.verbosity > 0:
if args.verbosity >= 2:
setup_log_func(alog, level=logging.DEBUG)
else:
setup_log_func(alog, level=logging.INFO)
elif hasattr(args, 'debug') and args.debug:
setup_log_func(alog, level=logging.DEBUG)
else:
alog.addHandler(logging.NullHandler())
log.debug(args)
alog.info("Starting tool version {v}".format(v=parser.version))
rcode = exe_runner_func(args)
run_time = time.time() - started_at
_d = dict(r=rcode, s=run_time)
alog.info("exiting with return code {r} in {s:.2f} sec.".format(**_d))
return rcode
def main_runner_default(argv, parser, alog):
# FIXME. This still has the old set_defaults(func=func) and
# has the assumption that --debug has been assigned as an option
# This is used for all the subparsers
setup_log_func = setup_log
return main_runner(argv, parser, args_executer, setup_log_func, alog)
|
