Source code for fastr.utils.cmd.trace

# Copyright 2011-2014 Biomedical Imaging Group Rotterdam, Departments of
# Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import os
import json
from fastr.helpers.report import print_job_result
from fastr.utils.cmd import add_parser_doc_link

FASTR_LOG_TYPE = 'none'


[docs]def get_parser(): parser = argparse.ArgumentParser( description="Fastr trace helps you inspect the staging directory of the" " Network run and pinpoint the errors." ) parser.add_argument('infile', nargs='?', default=os.path.join(os.getcwd(), '__sink_data__.json'), metavar='__sink_data__.json', help='result file to cat') parser.add_argument('--verbose', '-v', dest='verbose', action='store_true', help='set verbose output for more details') parser.add_argument('--sinks', dest='sinks', nargs="*", help='list results for specified sinks') parser.add_argument('--samples', dest='samples', nargs="*", help='list result for all samples') return parser
[docs]def read_sink_data(infile): if os.path.isdir(infile): infile = os.path.join(infile, '__sink_data__.json') if os.path.isfile(infile) and os.access(infile, os.R_OK): with open(infile) as sink_data_file: data = json.load(sink_data_file) else: print('ERROR: could not read: {}'.format(infile)) data = None return data
[docs]def switch_sample_sink(sink_data): # Switch samples and sinks samples_overview = dict() for sink, samples in sink_data.items(): for sample_id, sample in samples.items(): if sample_id not in samples_overview: samples_overview[sample_id] = dict() samples_overview[sample_id][sink] = sample return samples_overview
[docs]def main(): """ Trace samples/sinks from a run """ # No arguments were parsed yet, parse them now parser = add_parser_doc_link(get_parser(), __file__) args = parser.parse_args() sink_data = read_sink_data(args.infile) if not sink_data: exit(1) if args.sinks is not None and args.samples is not None: if len(args.sinks) == len(args.samples): sample_sink_tuples = list(zip(args.sinks, args.samples)) print_sample_sink(sink_data, os.path.dirname(args.infile), sample_sink_tuples, args.verbose) else: print("ERROR nr of sinks does not match number of samples") exit(1) elif args.sinks is not None: print_sinks(sink_data, args.sinks, args.verbose) elif args.samples is not None: print_samples(sink_data, args.samples, args.verbose) else: parser.print_help()
if __name__ == '__main__': main()