def get_parameter_type()

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

• 38 lines of code
• 22 McCabe index (conditional complexity)

def get_parameter_type(sample_input_ex, sample_output_ex=None):
    if sample_input_ex is None:
        input_param = NoSampleParameterType()
    else:
        try:
            schema = _infer_schema(sample_input_ex)
            schema_types = schema.input_types
        except MlflowException:
            pass
        finally:
            if isinstance(sample_input_ex, np.ndarray):
                # Unnamed tensor input
                input_param = NumpyParameterType(sample_input_ex, enforce_shape=False)
            elif pandas_installed and isinstance(sample_input_ex, pd.DataFrame):
                # TODO check with OSS about pd.Series
                input_param = PandasParameterType(
                    sample_input_ex, enforce_shape=False, orient="split"
                )
            # elif schema_types and isinstance(sample_input_ex, dict) and not all(stype == DataType.string for stype in schema_types) and \
            #     all(isinstance(value, list) for value in sample_input_ex.values()):
            #     # for dictionaries where there is any non-string type, named tensor
            #     param_arg = {}
            #     for key, value in sample_input_ex.items():
            #         param_arg[key] = NumpyParameterType(value, enforce_shape=False)
            #     input_param = StandardPythonParameterType(param_arg)
            elif isinstance(sample_input_ex, dict):
                # TODO keeping this around while _infer_schema doesn't work on dataframe string signatures
                param_arg = {}
                for key, value in sample_input_ex.items():
                    param_arg[key] = NumpyParameterType(value, enforce_shape=False)
                input_param = StandardPythonParameterType(param_arg)
            else:
                # strings, bytes, lists and dictionaries with only strings as base type
                input_param = NoSampleParameterType()

    if sample_output_ex is None:
        output_param = NoSampleParameterType()
    else:
        if isinstance(sample_output_ex, np.ndarray):
            # Unnamed tensor input
            output_param = NumpyParameterType(sample_output_ex, enforce_shape=False)
        elif isinstance(sample_output_ex, dict):
            param_arg = {}
            for key, value in sample_output_ex.items():
                param_arg[key] = NumpyParameterType(value, enforce_shape=False)
            output_param = StandardPythonParameterType(param_arg)
        else:
            output_param = PandasParameterType(
                sample_output_ex, enforce_shape=False, orient="records"
            )

    return input_param, output_param