in pai/serializers.py [0:0]
def serialize(self, data: Union[Dict[str, Any], tf_pb.PredictRequest]) -> bytes:
if isinstance(data, tf_pb.PredictRequest):
return data.SerializeToString()
request = tf_pb.PredictRequest()
if self._output_filter:
for output_name in self._output_filter:
request.output_filter.append(output_name)
if not isinstance(data, dict):
if not self._input_specs or len(self._input_specs) > 1:
raise ValueError(
"TensorFlowSerializer accepts a dictionary as input data, "
"with each input value having a name."
)
else:
# TensorFlow Processor expects key-value pairs for input data. However,
# if the input data is not a dictionary and the deployed model accepts
# exactly one input (by model signature), the key will be inferred
# from model signature and the current input data will be taken as
# value.
value = numpy.asarray(data)
input_spec = self._input_specs[0]
if (
input_spec.shape
and len([dim for dim in input_spec.shape if dim == -1]) == 1
):
value = value.reshape(input_spec.shape)
data_type = (
input_spec.data_type
if input_spec and input_spec.data_type is not None
else self._np_dtype_to_tf_dtype(value.dtype.type)
)
self._put_value(
request=request,
name=input_spec.name,
data_type=data_type,
shape=value.shape,
data=np.ravel(value).tolist(),
)
else:
input_specs_dict = (
{input_spec.name: input_spec for input_spec in self._input_specs}
if self._input_specs
else {}
)
for name, value in data.items():
input_spec = input_specs_dict.get(name)
if not isinstance(value, np.ndarray):
value = np.asarray(value)
data_type = (
input_spec.data_type
if input_spec and input_spec.data_type is not None
else self._np_dtype_to_tf_dtype(value.dtype.type)
)
if (
input_spec
and input_spec.shape
and len([dim for dim in input_spec.shape if dim == -1]) == 1
):
value = value.reshape(input_spec.shape)
self._put_value(
request=request,
name=input_spec.name,
data_type=data_type,
shape=value.shape,
data=np.ravel(value).tolist(),
)
return request.SerializeToString()