Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.tools.parameters.wrapped_json

import logging

log = logging.getLogger(__name__)

SKIP_INPUT = object()


[docs]def json_wrap(inputs, input_values, as_dict=None, handle_files="skip"): if as_dict is None: as_dict = {} for input in inputs.values(): input_name = input.name value = input_values[input_name] json_value = _json_wrap_input(input, value, handle_files=handle_files) if json_value is SKIP_INPUT: continue as_dict[input_name] = json_value return as_dict
def _json_wrap_input(input, value, handle_files="skip"): input_type = input.type def _data_input_to_path(v): path = _cast_if_not_none(v, str) if path == "None": path = None return path if input_type == "repeat": repeat_job_value = [] for d in value: repeat_instance_job_value = {} json_wrap(input.inputs, d, repeat_instance_job_value) repeat_job_value.append(repeat_instance_job_value) json_value = repeat_job_value elif input_type == "conditional": values = value current = values["__current_case__"] conditional_job_value = {} json_wrap(input.cases[current].inputs, values, conditional_job_value) test_param = input.test_param test_param_name = test_param.name test_value = _json_wrap_input(test_param, values[test_param_name]) conditional_job_value[test_param_name] = test_value json_value = conditional_job_value elif input_type == "section": values = value section_job_value = {} json_wrap(input.inputs, values, section_job_value) json_value = section_job_value elif input_type == "data" and input.multiple: if handle_files == "paths": json_value = list(map(_data_input_to_path, value)) elif handle_files == "skip": return SKIP_INPUT else: raise NotImplementedError() elif input_type == "data": if handle_files == "paths": json_value = _data_input_to_path(value) elif handle_files == "skip": return SKIP_INPUT elif handle_files == "OBJECT": if value: if isinstance(value, list): value = value[0] json_value = _hda_to_object(value) if input.load_contents: with open(str(value), mode='rb') as fh: json_value['contents'] = fh.read(input.load_contents).decode('utf-8', errors='replace') return json_value else: return None else: raise NotImplementedError() elif input_type == "data_collection": if handle_files == "skip": return SKIP_INPUT raise NotImplementedError() elif input_type in ["select", "text", "color", "hidden"]: json_value = _cast_if_not_none(value, str) elif input_type == "float": json_value = _cast_if_not_none(value, float, empty_to_none=True) elif input_type == "integer": json_value = _cast_if_not_none(value, int, empty_to_none=True) elif input_type == "boolean": json_value = _cast_if_not_none(value, bool) elif input_type == "data_column": # value is a SelectToolParameterWrapper() if input.multiple: json_value = [int(_) for _ in _cast_if_not_none(value.value, list)] else: json_value = [_cast_if_not_none(value.value, int)] else: raise NotImplementedError("input_type [%s] not implemented" % input_type) return json_value def _hda_to_object(hda): hda_dict = hda.to_dict() metadata_dict = {} for key, value in hda_dict.items(): if key.startswith("metadata_"): metadata_dict[key[len("metadata_"):]] = value return { 'file_ext': hda_dict['file_ext'], 'name': hda_dict['name'], 'metadata': metadata_dict, } def _cast_if_not_none(value, cast_to, empty_to_none=False): # log.debug("value [%s], type[%s]" % (value, type(value))) if value is None or (empty_to_none and str(value) == ''): return None else: return cast_to(value) __all__ = ('json_wrap', )