This document is for an old release 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', )