def parse_model_input_from_input_data_transformers()

in sdk/python/foundation-models/system/inference/text-to-image/scoring-files/score_online.py [0:0]

• 52 lines of code
• 30 McCabe index (conditional complexity)

def parse_model_input_from_input_data_transformers(input_data):
    # Format input
    if isinstance(input_data, str):
        try:
            input_data = json.loads(input_data)
        except ValueError:
            # allow non-json strings to go through
            input = input_data

    if isinstance(input_data, dict) and "input_data" in input_data:
        input_data = input_data["input_data"]

    if is_hfv2:
        input = input_data
    elif isinstance(input_data, str) or isinstance(input_data, bytes):
        # strings and bytes go through
        input = input_data
    elif isinstance(input_data, list) and all(
        isinstance(element, str) for element in input_data
    ):
        # lists of strings go through
        input = input_data
    elif isinstance(input_data, list) and all(
        isinstance(element, dict) for element in input_data
    ):
        # lists of dicts of [str: str | List[str]] go through
        try:
            for dict_input in input_data:
                _validate_input_dictionary_contains_only_strings_and_lists_of_strings(
                    dict_input
                )
            input = input_data
        except MlflowException:
            _logger.error(
                "Could not parse model input - passed a list of dictionaries which had entries which were not strings or lists."
            )
    elif isinstance(input_data, list):
        # if a list, assume the input is a numpy array
        input = np.asarray(input_data)
    elif (
        isinstance(input_data, dict)
        and "columns" in input_data
        and "index" in input_data
        and "data" in input_data
    ):
        # if the dictionary follows pandas split column format, deserialize into a pandas Dataframe
        input = pd.read_json(json.dumps(input_data), orient="split", dtype=False)
    elif isinstance(input_data, dict):
        # if input is a dictionary, but is not all ndarrays and is not pandas, it must only contain strings
        try:
            _validate_input_dictionary_contains_only_strings_and_lists_of_strings(
                input_data
            )
            input = input_data
        except MlflowException:
            # otherwise, assume input is a named tensor, and deserialize into a dict[str, numpy.ndarray]
            input = {
                input_name: np.asarray(input_value)
                for input_name, input_value in input_data.items()
            }
    else:
        input = input_data

    return input