    async def from_proto(self, field: pb.NDArray | bytes) -> ext.NpNDArray:
        """
        Process incoming protobuf request and convert it to ``numpy.ndarray``

        Args:
            request: Incoming RPC request message.
            context: grpc.ServicerContext

        Returns:
            ``numpy.ndarray``: A ``np.array`` constructed from given protobuf message.

        .. seealso::

            :ref:`Protobuf representation of np.ndarray <guides/grpc:Array representation via \\`\\`NDArray\\`\\`>`

        .. note::

           Currently, we support ``pb.NDArray`` and ``serialized_bytes`` as valid inputs.
           ``serialized_bytes`` will be prioritised over ``pb.NDArray`` if both are provided.
           Serialized bytes has a specialized bytes representation and should not be used by users directly.
        """
        if isinstance(field, bytes):
            # We will be using ``np.frombuffer`` to deserialize the bytes.
            # This means that we need to ensure that ``dtype`` are provided to the IO descriptor
            #
            # ```python
            # from __future__ import annotations
            #
            # import numpy as np
            #
            # @svc.api(input=NumpyNdarray(dtype=np.float16), output=NumpyNdarray())
            # def predict(input: np.ndarray):
            #     ... # input will be serialized with np.frombuffer, and hence dtype is required
            # ```
            if not self._dtype:
                raise BadInput(
                    "'serialized_bytes' requires specifying 'dtype'."
                ) from None

            dtype: ext.NpDTypeLike = self._dtype
            array = np.frombuffer(field, dtype=self._dtype)
        else:
            if isinstance(field, pb_v1alpha1.NDArray):
                version = "v1alpha1"
            elif isinstance(field, pb.NDArray):
                version = "v1"
            else:
                raise BadInput(
                    f"Expected 'pb.NDArray' or 'bytes' but received {type(field)}"
                ) from None

            # The behaviour of dtype are as follows:
            # - if not provided:
            #     * All of the fields are empty, then we return a ``np.empty``.
            #     * We will loop through all of the provided fields, and only allows one field per message.
            #       If here are more than two fields (i.e. ``string_values`` and ``float_values``), then we will raise an error, as we don't know how to deserialize the data.
            # - if provided:
            #     * We will use the provided dtype-to-field maps to get the data from the given message.
            if field.dtype == pb.NDArray.DTYPE_UNSPECIFIED:
                dtype = None
            else:
                try:
                    dtype = dtypepb_to_npdtype_map(version=version)[field.dtype]
                except KeyError:
                    raise BadInput(f"{field.dtype} is invalid.") from None
            if dtype is not None:
                values_array = getattr(
                    field, dtypepb_to_fieldpb_map(version=version)[field.dtype]
                )
            else:
                fieldpb = [
                    f.name for f, _ in field.ListFields() if f.name.endswith("_values")
                ]
                if len(fieldpb) == 0:
                    # input message doesn't have any fields.
                    return np.empty(shape=field.shape or 0)
                elif len(fieldpb) > 1:
                    # when there are more than two values provided in the proto.
                    raise BadInput(
                        f"Array contents can only be one of given values key. Use one of '{fieldpb}' instead.",
                    ) from None

                dtype: ext.NpDTypeLike = fieldpb_to_npdtype_map()[fieldpb[0]]
                values_array = getattr(field, fieldpb[0])
            try:
                array = np.array(values_array, dtype=dtype)
            except ValueError:
                array = np.array(values_array)

            # We will try to reshape the array if ``shape`` is provided.
            # Note that all of the logics here are handled in-place, meaning that we will ensure
            # not to create new copies of given initialized array.
            if field.shape:
                array = np.reshape(array, field.shape)

        # We will try to run validation process before sending this of to the user.
        return self.validate_array(array)
Process incoming protobuf request and convert it to ``numpy.ndarray``

Args:
request: Incoming RPC request message.
context: grpc.ServicerContext

Returns:
``numpy.ndarray``: A ``np.array`` constructed from given protobuf message.

.. seealso::

:ref:`Protobuf representation of np.ndarray <guides/grpc:Array representation via \\`\\`NDArray\\`\\`>`

.. note::

Currently, we support ``pb.NDArray`` and ``serialized_bytes`` as valid inputs.
``serialized_bytes`` will be prioritised over ``pb.NDArray`` if both are provided.
Serialized bytes has a specialized bytes representation and should not be used by users directly.
src/bentoml/_internal/io_descriptors/numpy.py