text: # coding=utf-8
  # 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.
  """
   Feature extraction saving/loading class for common feature extractors.
  """
  
  import copy
  import json
  import os
  from collections import UserDict
  from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
  
  import numpy as np
  
  from .file_utils import (
      FEATURE_EXTRACTOR_NAME,
      TensorType,
      _is_jax,
      _is_numpy,
      _is_torch_device,
      cached_path,
      hf_bucket_url,
      is_flax_available,
      is_offline_mode,
      is_remote_url,
      is_tf_available,
      is_torch_available,
      torch_required,
  )
  from .utils import logging
  
  
  if TYPE_CHECKING:
      if is_torch_available():
          import torch
  
  
  logger = logging.get_logger(__name__)
  
  PreTrainedFeatureExtractor = Union["SequenceFeatureExtractor"]  # noqa: F821
  
  
  class BatchFeature(UserDict):
      r"""
      Holds the output of the [`~SequenceFeatureExtractor.pad`] and 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 PyTorch/TensorFlow/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) -> Union[Any]:
          """
          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"]
  
      # Copied from transformers.tokenization_utils_base.BatchEncoding.keys
      def keys(self):
          return self.data.keys()
  
      # Copied from transformers.tokenization_utils_base.BatchEncoding.values
      def values(self):
          return self.data.values()
  
      # Copied from transformers.tokenization_utils_base.BatchEncoding.items
      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 [`~file_utils.TensorType`], *optional*):
                  The type of tensors to use. If `str`, should be one of the values of the enum
                  [`~file_utils.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.TENSORFLOW:
              if not is_tf_available():
                  raise ImportError(
                      "Unable to convert output to TensorFlow tensors format, TensorFlow is not installed."
                  )
              import tensorflow as tf
  
              as_tensor = tf.constant
              is_tensor = tf.is_tensor
          elif tensor_type == TensorType.PYTORCH:
              if not is_torch_available():
                  raise ImportError("Unable to convert output to PyTorch tensors format, PyTorch is not installed.")
              import torch
  
              def as_tensor(value):
                  if isinstance(value, (list, tuple)) and len(value) > 0 and isinstance(value[0], np.ndarray):
                      value = np.array(value)
                  return torch.tensor(value)
  
              is_tensor = torch.is_tensor
          elif tensor_type == TensorType.JAX:
              if not is_flax_available():
                  raise ImportError("Unable to convert output to JAX tensors format, JAX is not installed.")
              import jax.numpy as jnp  # noqa: F811
  
              as_tensor = jnp.array
              is_tensor = _is_jax
          else:
              as_tensor = np.asarray
              is_tensor = _is_numpy
  
          # 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_values":
                      raise ValueError("Unable to create tensor returning overflowing values of different lengths. ")
                  raise ValueError(
                      "Unable to create tensor, you should probably activate padding "
                      "with 'padding=True' to have batched tensors with the same length."
                  )
  
          return self
  
      @torch_required
      # Copied from transformers.tokenization_utils_base.BatchEncoding.to with BatchEncoding->BatchFeature
      def to(self, device: Union[str, "torch.device"]) -> "BatchFeature":
          """
          Send all values to device by calling `v.to(device)` (PyTorch only).
  
          Args:
              device (`str` or `torch.device`): The device to put the tensors on.
  
          Returns:
              [`BatchFeature`]: The same instance after modification.
          """
  
          # This check catches things like APEX blindly calling "to" on all inputs to a module
          # Otherwise it passes the casts down and casts the LongTensor containing the token idxs
          # into a HalfTensor
          if isinstance(device, str) or _is_torch_device(device) or isinstance(device, int):
              self.data = {k: v.to(device=device) for k, v in self.data.items()}
          else:
              logger.warning(f"Attempting to cast a BatchFeature to type {str(device)}. This is not supported.")
          return self
  
  
  class FeatureExtractionMixin:
      """
      This is a feature extraction mixin used to provide saving/loading functionality for sequential and image feature
      extractors.
      """
  
      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
      ) -> PreTrainedFeatureExtractor:
          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 *model id* of a pretrained feature_extractor hosted inside a model repo on
                    huggingface.co. Valid model ids can be located at the root-level, like `bert-base-uncased`, or
                    namespaced under a user or organization name, like `dbmdz/bert-base-german-cased`.
                  - 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`.
              cache_dir (`str` or `os.PathLike`, *optional*):
                  Path to a directory in which a downloaded pretrained model feature extractor should be cached if the
                  standard cache should not be used.
              force_download (`bool`, *optional*, defaults to `False`):
                  Whether or not to force to (re-)download the feature extractor files and override the cached versions
                  if they exist.
              resume_download (`bool`, *optional*, defaults to `False`):
                  Whether or not to delete incompletely received file. Attempts to resume the download if such a file
                  exists.
              proxies (`Dict[str, str]`, *optional*):
                  A dictionary of proxy servers to use by protocol or endpoint, e.g., `{'http': 'foo.bar:3128',
                  'http://hostname': 'foo.bar:4012'}.` The proxies are used on each request.
              use_auth_token (`str` or *bool*, *optional*):
                  The token to use as HTTP bearer authorization for remote files. If `True`, will use the token generated
                  when running `transformers-cli login` (stored in `~/.huggingface`).
              revision(`str`, *optional*, defaults to `"main"`):
                  The specific model version to use. It can be a branch name, a tag name, or a commit id, since we use a
                  git-based system for storing models and other artifacts on huggingface.co, so `revision` can be any
                  identifier allowed by git.
              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.
  
          <Tip>
  
          Passing `use_auth_token=True` is required when you want to use a private model.
  
          </Tip>
  
          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: *Wav2Vec2FeatureExtractor*
          feature_extractor = Wav2Vec2FeatureExtractor.from_pretrained(
              "facebook/wav2vec2-base-960h"
          )  # Download feature_extraction_config from huggingface.co and cache.
          feature_extractor = Wav2Vec2FeatureExtractor.from_pretrained(
              "./test/saved_model/"
          )  # E.g. feature_extractor (or model) was saved using *save_pretrained('./test/saved_model/')*
          feature_extractor = Wav2Vec2FeatureExtractor.from_pretrained("./test/saved_model/preprocessor_config.json")
          feature_extractor = Wav2Vec2FeatureExtractor.from_pretrained(
              "facebook/wav2vec2-base-960h", return_attention_mask=False, foo=False
          )
          assert feature_extractor.return_attention_mask is False
          feature_extractor, unused_kwargs = Wav2Vec2FeatureExtractor.from_pretrained(
              "facebook/wav2vec2-base-960h", return_attention_mask=False, foo=False, return_unused_kwargs=True
          )
          assert feature_extractor.return_attention_mask is False
          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]):
          """
          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).
          """
          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"Configuration saved in {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)
          force_download = kwargs.pop("force_download", False)
          resume_download = kwargs.pop("resume_download", False)
          proxies = kwargs.pop("proxies", None)
          use_auth_token = kwargs.pop("use_auth_token", None)
          local_files_only = kwargs.pop("local_files_only", False)
          revision = kwargs.pop("revision", None)
  
          from_pipeline = kwargs.pop("_from_pipeline", None)
          from_auto_class = kwargs.pop("_from_auto", False)
  
          user_agent = {"file_type": "feature extractor", "from_auto_class": from_auto_class}
          if from_pipeline is not None:
              user_agent["using_pipeline"] = from_pipeline
  
          if is_offline_mode() and not local_files_only:
              logger.info("Offline mode: forcing local_files_only=True")
              local_files_only = True
  
          pretrained_model_name_or_path = str(pretrained_model_name_or_path)
          if os.path.isdir(pretrained_model_name_or_path):
              feature_extractor_file = os.path.join(pretrained_model_name_or_path, FEATURE_EXTRACTOR_NAME)
          elif os.path.isfile(pretrained_model_name_or_path) or is_remote_url(pretrained_model_name_or_path):
              feature_extractor_file = pretrained_model_name_or_path
          else:
              feature_extractor_file = hf_bucket_url(
                  pretrained_model_name_or_path, filename=FEATURE_EXTRACTOR_NAME, revision=revision, mirror=None
              )
  
          try:
              # Load from URL or cache if already cached
              resolved_feature_extractor_file = cached_path(
                  feature_extractor_file,
                  cache_dir=cache_dir,
                  force_download=force_download,
                  proxies=proxies,
                  resume_download=resume_download,
                  local_files_only=local_files_only,
                  use_auth_token=use_auth_token,
                  user_agent=user_agent,
              )
              # 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 EnvironmentError as err:
              logger.error(err)
              msg = (
                  f"Can't load feature extractor for '{pretrained_model_name_or_path}'. Make sure that:\n\n"
                  f"- '{pretrained_model_name_or_path}' is a correct model identifier listed on 'https://huggingface.co/models'\n"
                  f"  (make sure '{pretrained_model_name_or_path}' is not a path to a local directory with something else, in that case)\n\n"
                  f"- or '{pretrained_model_name_or_path}' is the correct path to a directory containing a {FEATURE_EXTRACTOR_NAME} file\n\n"
              )
              raise EnvironmentError(msg)
  
          except json.JSONDecodeError:
              msg = (
                  f"Couldn't reach server at '{feature_extractor_file}' to download feature extractor configuration file or "
                  "feature extractor configuration file is not a valid JSON file. "
                  f"Please check network or file content here: {resolved_feature_extractor_file}."
              )
              raise EnvironmentError(msg)
  
          if resolved_feature_extractor_file == feature_extractor_file:
              logger.info(f"loading feature extractor configuration file {feature_extractor_file}")
          else:
              logger.info(
                  f"loading feature extractor configuration file {feature_extractor_file} from cache at {resolved_feature_extractor_file}"
              )
  
          return feature_extractor_dict, kwargs
  
      @classmethod
      def from_dict(cls, feature_extractor_dict: Dict[str, Any], **kwargs) -> PreTrainedFeatureExtractor:
          """
          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)
  
          logger.info(f"Feature extractor {feature_extractor}")
          if return_unused_kwargs:
              return feature_extractor, kwargs
          else:
              return feature_extractor
  
      def to_dict(self) -> 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]) -> PreTrainedFeatureExtractor:
          """
          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()}"
  