wehub-resource-sync
379 lines
15 KiB
Python
379 lines
15 KiB
Python
# coding=utf-8
|
|
# Copyright (c) 2022 PaddlePaddle Authors. All Rights Reserved.
|
|
# Copyright 2021 The HuggingFace Inc. team.
|
|
#
|
|
# 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.
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
from collections import UserDict
|
|
from typing import Any, Dict, Optional, Tuple, Union
|
|
|
|
import numpy as np
|
|
import paddle
|
|
|
|
from paddlenlp.utils.download import resolve_file_path
|
|
|
|
from ..utils.log import logger
|
|
from .tokenizer_utils_base import TensorType
|
|
|
|
FEATURE_EXTRACTOR_NAME = "preprocessor_config.json"
|
|
|
|
|
|
class BatchFeature(UserDict):
|
|
r"""
|
|
Holds the feature extractor specific `__call__` methods.
|
|
This class is derived from a python dictionary and can be used as a dictionary.
|
|
Args:
|
|
data (`dict`):
|
|
Dictionary of lists/arrays/tensors returned by the __call__/pad methods ('input_values', 'attention_mask',
|
|
etc.).
|
|
tensor_type (`Union[None, str, TensorType]`, *optional*):
|
|
You can give a tensor_type here to convert the lists of integers in Paddle/Numpy Tensors at
|
|
initialization.
|
|
"""
|
|
|
|
def __init__(self, data: Optional[Dict[str, Any]] = None, tensor_type: Union[None, str, TensorType] = None):
|
|
super().__init__(data)
|
|
self.convert_to_tensors(tensor_type=tensor_type)
|
|
|
|
def __getitem__(self, item: str):
|
|
"""
|
|
If the key is a string, returns the value of the dict associated to `key` ('input_values', 'attention_mask',
|
|
etc.).
|
|
"""
|
|
if isinstance(item, str):
|
|
return self.data[item]
|
|
else:
|
|
raise KeyError("Indexing with integers is not available when using Python based feature extractors")
|
|
|
|
def __getattr__(self, item: str):
|
|
try:
|
|
return self.data[item]
|
|
except KeyError:
|
|
raise AttributeError
|
|
|
|
def __getstate__(self):
|
|
return {"data": self.data}
|
|
|
|
def __setstate__(self, state):
|
|
if "data" in state:
|
|
self.data = state["data"]
|
|
|
|
def keys(self):
|
|
return self.data.keys()
|
|
|
|
def values(self):
|
|
return self.data.values()
|
|
|
|
def items(self):
|
|
return self.data.items()
|
|
|
|
def convert_to_tensors(self, tensor_type: Optional[Union[str, TensorType]] = None):
|
|
"""
|
|
Convert the inner content to tensors.
|
|
Args:
|
|
tensor_type (`str` or [`TensorType`], *optional*):
|
|
The type of tensors to use. If `str`, should be one of the values of the enum [`TensorType`]. If
|
|
`None`, no modification is done.
|
|
"""
|
|
if tensor_type is None:
|
|
return self
|
|
|
|
# Convert to TensorType
|
|
if not isinstance(tensor_type, TensorType):
|
|
tensor_type = TensorType(tensor_type)
|
|
|
|
# Get a function reference for the correct framework
|
|
if tensor_type == TensorType.PADDLE:
|
|
as_tensor = paddle.to_tensor
|
|
is_tensor = paddle.is_tensor
|
|
else:
|
|
as_tensor = np.asarray
|
|
|
|
def is_tensor(x):
|
|
return isinstance(x, np.ndarray)
|
|
|
|
# Do the tensor conversion in batch
|
|
for key, value in self.items():
|
|
try:
|
|
if not is_tensor(value):
|
|
tensor = as_tensor(value)
|
|
|
|
self[key] = tensor
|
|
except: # noqa E722
|
|
if key == "overflowing_tokens":
|
|
raise ValueError(
|
|
"Unable to create tensor returning overflowing tokens of different lengths. "
|
|
"Please see if a fast version of this tokenizer is available to have this feature available."
|
|
)
|
|
raise ValueError(
|
|
"Unable to create tensor, you should probably activate truncation and/or padding "
|
|
"with 'padding=True' 'truncation=True' to have batched tensors with the same length."
|
|
)
|
|
|
|
return self
|
|
|
|
|
|
class FeatureExtractionMixin(object):
|
|
"""
|
|
This is a feature extraction mixin used to provide saving/loading functionality for sequential and image feature
|
|
extractors.
|
|
"""
|
|
|
|
pretrained_init_configuration = {}
|
|
|
|
pretrained_feature_extractor_file = []
|
|
_auto_class = None
|
|
|
|
def __init__(self, **kwargs):
|
|
"""Set elements of `kwargs` as attributes."""
|
|
# Pop "processor_class" as it should be saved as private attribute
|
|
self._processor_class = kwargs.pop("processor_class", None)
|
|
# Additional attributes without default values
|
|
for key, value in kwargs.items():
|
|
try:
|
|
setattr(self, key, value)
|
|
except AttributeError as err:
|
|
logger.error(f"Can't set {key} with value {value} for {self}")
|
|
raise err
|
|
|
|
def _set_processor_class(self, processor_class: str):
|
|
"""Sets processor class as an attribute."""
|
|
self._processor_class = processor_class
|
|
|
|
@classmethod
|
|
def from_pretrained(cls, pretrained_model_name_or_path: Union[str, os.PathLike], **kwargs):
|
|
r"""
|
|
Instantiate a type of [`~feature_extraction_utils.FeatureExtractionMixin`] from a feature extractor, *e.g.* a
|
|
derived class of [`SequenceFeatureExtractor`].
|
|
|
|
Args:
|
|
pretrained_model_name_or_path (`str` or `os.PathLike`):
|
|
This can be either:
|
|
|
|
- a string, the name of a community-contributed pretrained or built-in pretrained model.
|
|
- a path to a *directory* containing a feature extractor file saved using the
|
|
[`~feature_extraction_utils.FeatureExtractionMixin.save_pretrained`] method, e.g.,
|
|
`./my_model_directory/`.
|
|
- a path or url to a saved feature extractor JSON *file*, e.g.,
|
|
`./my_model_directory/preprocessor_config.json`.
|
|
return_unused_kwargs (`bool`, *optional*, defaults to `False`):
|
|
If `False`, then this function returns just the final feature extractor object. If `True`, then this
|
|
functions returns a `Tuple(feature_extractor, unused_kwargs)` where *unused_kwargs* is a dictionary
|
|
consisting of the key/value pairs whose keys are not feature extractor attributes: i.e., the part of
|
|
`kwargs` which has not been used to update `feature_extractor` and is otherwise ignored.
|
|
kwargs (`Dict[str, Any]`, *optional*):
|
|
The values in kwargs of any keys which are feature extractor attributes will be used to override the
|
|
loaded values. Behavior concerning key/value pairs whose keys are *not* feature extractor attributes is
|
|
controlled by the `return_unused_kwargs` keyword parameter.
|
|
|
|
Returns:
|
|
A feature extractor of type [`~feature_extraction_utils.FeatureExtractionMixin`].
|
|
|
|
Examples:
|
|
|
|
```python
|
|
# We can't instantiate directly the base class *FeatureExtractionMixin* nor *SequenceFeatureExtractor* so let's show the examples on a
|
|
# derived class: *CLIPFeatureExtractor*
|
|
feature_extractor = CLIPFeatureExtractor.from_pretrained(
|
|
"openai/clip-vit-base-patch32"
|
|
) # Download feature_extraction_config from bos and cache.
|
|
feature_extractor = CLIPFeatureExtractor.from_pretrained(
|
|
"./test/saved_model/"
|
|
) # E.g. feature_extractor (or model) was saved using *save_pretrained('./test/saved_model/')*
|
|
feature_extractor = CLIPFeatureExtractor.from_pretrained("./test/saved_model/preprocessor_config.json")
|
|
feature_extractor, unused_kwargs = CLIPFeatureExtractor.from_pretrained(
|
|
"openai/clip-vit-base-patch32", foo=False, return_unused_kwargs=True
|
|
)
|
|
assert unused_kwargs == {"foo": False}
|
|
```
|
|
"""
|
|
feature_extractor_dict, kwargs = cls.get_feature_extractor_dict(pretrained_model_name_or_path, **kwargs)
|
|
|
|
return cls.from_dict(feature_extractor_dict, **kwargs)
|
|
|
|
def save_pretrained(self, save_directory: Union[str, os.PathLike], **kwargs):
|
|
"""
|
|
Save a feature_extractor object to the directory `save_directory`, so that it can be re-loaded using the
|
|
[`~feature_extraction_utils.FeatureExtractionMixin.from_pretrained`] class method.
|
|
|
|
Args:
|
|
save_directory (`str` or `os.PathLike`):
|
|
Directory where the feature extractor JSON file will be saved (will be created if it does not exist).
|
|
kwargs:
|
|
Additional key word arguments.
|
|
"""
|
|
if os.path.isfile(save_directory):
|
|
raise AssertionError(f"Provided path ({save_directory}) should be a directory, not a file")
|
|
|
|
os.makedirs(save_directory, exist_ok=True)
|
|
|
|
# If we save using the predefined names, we can load using `from_pretrained`
|
|
output_feature_extractor_file = os.path.join(save_directory, FEATURE_EXTRACTOR_NAME)
|
|
|
|
self.to_json_file(output_feature_extractor_file)
|
|
logger.info(f"Feature extractor saved in {output_feature_extractor_file}")
|
|
|
|
return [output_feature_extractor_file]
|
|
|
|
@classmethod
|
|
def get_feature_extractor_dict(
|
|
cls, pretrained_model_name_or_path: Union[str, os.PathLike], **kwargs
|
|
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
|
|
"""
|
|
From a `pretrained_model_name_or_path`, resolve to a dictionary of parameters, to be used for instantiating a
|
|
feature extractor of type [`~feature_extraction_utils.FeatureExtractionMixin`] using `from_dict`.
|
|
|
|
Parameters:
|
|
pretrained_model_name_or_path (`str` or `os.PathLike`):
|
|
The identifier of the pre-trained checkpoint from which we want the dictionary of parameters.
|
|
|
|
Returns:
|
|
`Tuple[Dict, Dict]`: The dictionary(ies) that will be used to instantiate the feature extractor object.
|
|
"""
|
|
cache_dir = kwargs.pop("cache_dir", None)
|
|
from_hf_hub = kwargs.pop("from_hf_hub", False)
|
|
from_aistudio = kwargs.pop("from_aistudio", False)
|
|
subfolder = kwargs.pop("subfolder", "")
|
|
if subfolder is None:
|
|
subfolder = ""
|
|
|
|
pretrained_model_name_or_path = str(pretrained_model_name_or_path)
|
|
resolved_feature_extractor_file = resolve_file_path(
|
|
pretrained_model_name_or_path,
|
|
[FEATURE_EXTRACTOR_NAME],
|
|
subfolder,
|
|
cache_dir=cache_dir,
|
|
from_aistudio=from_aistudio,
|
|
from_hf_hub=from_hf_hub,
|
|
)
|
|
assert (
|
|
resolved_feature_extractor_file is not None
|
|
), f"please make sure {FEATURE_EXTRACTOR_NAME} under {pretrained_model_name_or_path}"
|
|
try:
|
|
# Load feature_extractor dict
|
|
with open(resolved_feature_extractor_file, "r", encoding="utf-8") as reader:
|
|
text = reader.read()
|
|
feature_extractor_dict = json.loads(text)
|
|
|
|
except json.JSONDecodeError:
|
|
raise EnvironmentError(
|
|
f"It looks like the config file at '{resolved_feature_extractor_file}' is not a valid JSON file."
|
|
)
|
|
|
|
return feature_extractor_dict, kwargs
|
|
|
|
@classmethod
|
|
def from_dict(cls, feature_extractor_dict: Dict[str, Any], **kwargs):
|
|
"""
|
|
Instantiates a type of [`~feature_extraction_utils.FeatureExtractionMixin`] from a Python dictionary of
|
|
parameters.
|
|
|
|
Args:
|
|
feature_extractor_dict (`Dict[str, Any]`):
|
|
Dictionary that will be used to instantiate the feature extractor object. Such a dictionary can be
|
|
retrieved from a pretrained checkpoint by leveraging the
|
|
[`~feature_extraction_utils.FeatureExtractionMixin.to_dict`] method.
|
|
kwargs (`Dict[str, Any]`):
|
|
Additional parameters from which to initialize the feature extractor object.
|
|
|
|
Returns:
|
|
[`~feature_extraction_utils.FeatureExtractionMixin`]: The feature extractor object instantiated from those
|
|
parameters.
|
|
"""
|
|
return_unused_kwargs = kwargs.pop("return_unused_kwargs", False)
|
|
|
|
feature_extractor = cls(**feature_extractor_dict)
|
|
|
|
# Update feature_extractor with kwargs if needed
|
|
to_remove = []
|
|
for key, value in kwargs.items():
|
|
if hasattr(feature_extractor, key):
|
|
setattr(feature_extractor, key, value)
|
|
to_remove.append(key)
|
|
for key in to_remove:
|
|
kwargs.pop(key, None)
|
|
|
|
if return_unused_kwargs:
|
|
return feature_extractor, kwargs
|
|
else:
|
|
return feature_extractor
|
|
|
|
def to_dict(self, *args, **kwargs) -> Dict[str, Any]:
|
|
"""
|
|
Serializes this instance to a Python dictionary.
|
|
|
|
Returns:
|
|
`Dict[str, Any]`: Dictionary of all the attributes that make up this feature extractor instance.
|
|
"""
|
|
output = copy.deepcopy(self.__dict__)
|
|
output["feature_extractor_type"] = self.__class__.__name__
|
|
|
|
return output
|
|
|
|
@classmethod
|
|
def from_json_file(cls, json_file: Union[str, os.PathLike]):
|
|
"""
|
|
Instantiates a feature extractor of type [`~feature_extraction_utils.FeatureExtractionMixin`] from the path to
|
|
a JSON file of parameters.
|
|
|
|
Args:
|
|
json_file (`str` or `os.PathLike`):
|
|
Path to the JSON file containing the parameters.
|
|
|
|
Returns:
|
|
A feature extractor of type [`~feature_extraction_utils.FeatureExtractionMixin`]: The feature_extractor
|
|
object instantiated from that JSON file.
|
|
"""
|
|
with open(json_file, "r", encoding="utf-8") as reader:
|
|
text = reader.read()
|
|
feature_extractor_dict = json.loads(text)
|
|
return cls(**feature_extractor_dict)
|
|
|
|
def to_json_string(self) -> str:
|
|
"""
|
|
Serializes this instance to a JSON string.
|
|
|
|
Returns:
|
|
`str`: String containing all the attributes that make up this feature_extractor instance in JSON format.
|
|
"""
|
|
dictionary = self.to_dict()
|
|
|
|
for key, value in dictionary.items():
|
|
if isinstance(value, np.ndarray):
|
|
dictionary[key] = value.tolist()
|
|
|
|
# make sure private name "_processor_class" is correctly
|
|
# saved as "processor_class"
|
|
_processor_class = dictionary.pop("_processor_class", None)
|
|
if _processor_class is not None:
|
|
dictionary["processor_class"] = _processor_class
|
|
|
|
return json.dumps(dictionary, indent=2, sort_keys=True) + "\n"
|
|
|
|
def to_json_file(self, json_file_path: Union[str, os.PathLike]):
|
|
"""
|
|
Save this instance to a JSON file.
|
|
|
|
Args:
|
|
json_file_path (`str` or `os.PathLike`):
|
|
Path to the JSON file in which this feature_extractor instance's parameters will be saved.
|
|
"""
|
|
with open(json_file_path, "w", encoding="utf-8") as writer:
|
|
writer.write(self.to_json_string())
|
|
|
|
def __repr__(self):
|
|
return f"{self.__class__.__name__} {self.to_json_string()}"
|