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 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 _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', )