paddlepaddle--paddle/python/paddle/inference/wrapper.py

#   Copyright (c) 2020 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import logging
import os
from typing import TYPE_CHECKING, Any, TypedDict

import numpy as np

import paddle
from paddle.base import core
from paddle.base.core import (
    AnalysisConfig,
    PaddleDType,
    PaddleInferPredictor,
    PaddleInferTensor,
    PaddlePlace,
    convert_to_mixed_precision_bind,
)
from paddle.base.log_helper import get_logger

if TYPE_CHECKING:
    import numpy.typing as npt
    from typing_extensions import Unpack

    from paddle import Tensor

    class _WhiteList(TypedDict):
        white_list: set[str]


_logger = get_logger(
    __name__, logging.INFO, fmt='%(asctime)s-%(levelname)s: %(message)s'
)

DataType = PaddleDType
PlaceType = PaddlePlace
PrecisionType = AnalysisConfig.Precision
Config = AnalysisConfig
Tensor = PaddleInferTensor
Predictor = PaddleInferPredictor


def tensor_copy_from_cpu(self, data: npt.NDArray[Any] | list[str]) -> None:
    '''
    Support input type check based on tensor.copy_from_cpu.
    '''
    if isinstance(data, np.ndarray) or (
        isinstance(data, list) and len(data) > 0 and isinstance(data[0], str)
    ):
        self._copy_from_cpu_bind(data)
    else:
        raise TypeError(
            "In copy_from_cpu, we only support numpy ndarray and list[str] data type."
        )


def tensor_share_external_data(self, data: Tensor) -> None:
    '''
    Support input type check based on tensor.share_external_data.
    '''
    if isinstance(data, core.DenseTensor):
        self._share_external_data_bind(data)
    elif isinstance(data, paddle.Tensor):
        self._share_external_data_paddle_tensor_bind(data)
    elif isinstance(data, paddle.base.framework.Variable):
        raise TypeError(
            "The interface 'share_external_data' can only be used in dynamic graph mode. "
            "Maybe you called 'paddle.enable_static()' and you are in static graph mode now. "
            "Please use 'copy_from_cpu' instead."
        )
    else:
        raise TypeError(
            "In share_external_data, we only support Tensor and DenseTensor."
        )


def tensor_share_external_data_by_ptr_name(self, data, shape, dtype, place):
    '''
    Support tensor.share_external_data_by_ptr_name.
    '''
    self._share_external_data_by_ptr_name_bind(data, shape, dtype, place)


def convert_to_mixed_precision(
    model_file: str,
    params_file: str,
    mixed_model_file: str,
    mixed_params_file: str,
    mixed_precision: PrecisionType,
    backend: PlaceType,
    keep_io_types: bool = True,
    black_list: set[str] = set(),
    **kwargs: Unpack[_WhiteList],
) -> None:
    '''
    Convert a fp32 model to mixed precision model.

    Args:
        model_file: fp32 model file, e.g. inference.pdmodel.
        params_file: fp32 params file, e.g. inference.pdiparams.
        mixed_model_file: The storage path of the converted mixed-precision model.
        mixed_params_file: The storage path of the converted mixed-precision params.
        mixed_precision: The precision, e.g. PrecisionType.Half.
        backend: The backend, e.g. PlaceType.GPU.
        keep_io_types: Whether the model input and output dtype remains unchanged.
            Default is True.
        black_list: Operators that do not convert precision.
        kwargs: Supported keys including 'white_list'.
            - white_list: Operators that do convert precision.
    '''
    if backend is PlaceType.GPU and not core.is_compiled_with_cuda():
        _logger.error(
            "You should use PaddlePaddle compiled with GPU when backend set to PlaceType.GPU"
        )

    mixed_model_dirname = os.path.dirname(mixed_model_file)
    # Support mixed_params_file is empty, because some models don't have params, but convert_to_mixed_precision will call
    # constant_folding_pass, it will generate a new params file to save persistable vars, which is saved in the same
    # level file directory as the model file by default and ends in pdiparams.
    mixed_params_dirname = (
        os.path.dirname(mixed_params_file)
        if len(mixed_params_file) != 0
        else mixed_model_dirname
    )
    if not os.path.exists(mixed_params_dirname):
        os.makedirs(mixed_params_dirname)
    white_list = kwargs.get('white_list', set())
    convert_to_mixed_precision_bind(
        model_file,
        params_file,
        mixed_model_file,
        mixed_params_file,
        mixed_precision,
        backend,
        keep_io_types,
        black_list,
        white_list,
    )


Tensor.copy_from_cpu = tensor_copy_from_cpu
Tensor.share_external_data = tensor_share_external_data
Tensor.share_external_data_by_ptr_name = tensor_share_external_data_by_ptr_name