


'''simple docstring'''






import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


sys.path.append(str(Path(__file__).parent.parent.parent.parent / "utils"))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


__A					:  int											=      get_tests_dir("fixtures")
__A					:  Union[str, Any]											=      get_tests_dir("fixtures/dummy_feature_extractor_config.json")
__A					:  List[str]											=      get_tests_dir("fixtures/dummy-config.json")






class    __snake_case       (   unittest.TestCase):







    """simple docstring"""



    def __lowercase   (   self    :  Optional[Any]		)   ->  int:
      lowerCAmelCase_			:      Any			=   0

    def __lowercase   (   self    :  List[Any]		)   ->  int:
      lowerCAmelCase_			:      Union[str, Any]			=   AutoFeatureExtractor.from_pretrained("""facebook/wav2vec2-base-960h"""		)
      self.assertIsInstance(lowerCamelCase    ,						lowerCamelCase		)

    def __lowercase   (   self    :  Tuple		)   ->  Dict:
      lowerCAmelCase_			:      Tuple			=   AutoFeatureExtractor.from_pretrained(lowerCamelCase		)
      self.assertIsInstance(lowerCamelCase    ,						lowerCamelCase		)

    def __lowercase   (   self    :  Optional[int]		)   ->  Any:
      with tempfile.TemporaryDirectory() as tmpdirname:
        lowerCAmelCase_			:      List[str]			=   WavaVecaConfig()

        # remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
        lowerCAmelCase_			:      Any			=   AutoFeatureExtractor.from_pretrained(lowerCamelCase		).to_dict()

        config_dict.pop("""feature_extractor_type"""		)
        lowerCAmelCase_			:      str			=   WavaVecaFeatureExtractor(**lowerCamelCase		)

        # save in new folder
        model_config.save_pretrained(lowerCamelCase		)
        config.save_pretrained(lowerCamelCase		)

        lowerCAmelCase_			:      Optional[int]			=   AutoFeatureExtractor.from_pretrained(lowerCamelCase		)

        # make sure private variable is not incorrectly saved
        lowerCAmelCase_			:      List[str]			=   json.loads(config.to_json_string()		)
        self.assertTrue("""_processor_class""" not in dict_as_saved		)

      self.assertIsInstance(lowerCamelCase    ,						lowerCamelCase		)

    def __lowercase   (   self    :  Tuple		)   ->  Tuple:
      lowerCAmelCase_			:      List[str]			=   AutoFeatureExtractor.from_pretrained(lowerCamelCase		)
      self.assertIsInstance(lowerCamelCase    ,						lowerCamelCase		)

    def __lowercase   (   self    :  Union[str, Any]		)   ->  Tuple:
      with self.assertRaisesRegex(
          lowerCamelCase    ,						"""bert-base is not a local folder and is not a valid model identifier"""		):
        lowerCAmelCase_			:      Dict			=   AutoFeatureExtractor.from_pretrained("""bert-base"""		)

    def __lowercase   (   self    :  Union[str, Any]		)   ->  Any:
      with self.assertRaisesRegex(
          lowerCamelCase    ,						R"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""		):
        lowerCAmelCase_			:      Tuple			=   AutoFeatureExtractor.from_pretrained(lowerCamelCase    ,						revision="""aaaaaa"""		)

    def __lowercase   (   self    :  Optional[int]		)   ->  Any:
      with self.assertRaisesRegex(
          lowerCamelCase    ,						"""hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."""    ,						):
        lowerCAmelCase_			:      Dict			=   AutoFeatureExtractor.from_pretrained("""hf-internal-testing/config-no-model"""		)

    def __lowercase   (   self    :  int		)   ->  Optional[Any]:
      # If remote code is not set, we will time out when asking whether to load the model.
      with self.assertRaises(lowerCamelCase		):
        lowerCAmelCase_			:      Optional[Any]			=   AutoFeatureExtractor.from_pretrained(
            """hf-internal-testing/test_dynamic_feature_extractor"""		)
      # If remote code is disabled, we can't load this config.
      with self.assertRaises(lowerCamelCase		):
        lowerCAmelCase_			:      int			=   AutoFeatureExtractor.from_pretrained(
            """hf-internal-testing/test_dynamic_feature_extractor"""    ,						trust_remote_code=lowerCamelCase		)

      lowerCAmelCase_			:      Optional[Any]			=   AutoFeatureExtractor.from_pretrained(
          """hf-internal-testing/test_dynamic_feature_extractor"""    ,						trust_remote_code=lowerCamelCase		)
      self.assertEqual(feature_extractor.__class__.__name__    ,						"""NewFeatureExtractor"""		)

      # Test feature extractor can be reloaded.
      with tempfile.TemporaryDirectory() as tmp_dir:
        feature_extractor.save_pretrained(lowerCamelCase		)
        lowerCAmelCase_			:      int			=   AutoFeatureExtractor.from_pretrained(lowerCamelCase    ,						trust_remote_code=lowerCamelCase		)
      self.assertEqual(reloaded_feature_extractor.__class__.__name__    ,						"""NewFeatureExtractor"""		)

    def __lowercase   (   self    :  Optional[Any]		)   ->  Optional[Any]:
      try:
        AutoConfig.register("""custom"""    ,						lowerCamelCase		)
        AutoFeatureExtractor.register(lowerCamelCase    ,						lowerCamelCase		)
        # Trying to register something existing in the Transformers library will raise an error
        with self.assertRaises(lowerCamelCase		):
          AutoFeatureExtractor.register(lowerCamelCase    ,						lowerCamelCase		)

        # Now that the config is registered, it can be used as any other config with the auto-API
        lowerCAmelCase_			:      List[str]			=   CustomFeatureExtractor.from_pretrained(lowerCamelCase		)
        with tempfile.TemporaryDirectory() as tmp_dir:
          feature_extractor.save_pretrained(lowerCamelCase		)
          lowerCAmelCase_			:      Union[str, Any]			=   AutoFeatureExtractor.from_pretrained(lowerCamelCase		)
          self.assertIsInstance(lowerCamelCase    ,						lowerCamelCase		)

      finally:
        if "custom" in CONFIG_MAPPING._extra_content:
          del CONFIG_MAPPING._extra_content["custom"]
        if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
          del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]





    def __lowercase   (   self    :  List[str]		)   ->  List[str]:






      class    __snake_case       (   _SCREAMING_SNAKE_CASE):







          """simple docstring"""



          lowercase    =					True

      try:
        AutoConfig.register("""custom"""    ,						lowerCamelCase		)
        AutoFeatureExtractor.register(lowerCamelCase    ,						lowerCamelCase		)
        # If remote code is not set, the default is to use local
        lowerCAmelCase_			:      str			=   AutoFeatureExtractor.from_pretrained(
            """hf-internal-testing/test_dynamic_feature_extractor"""		)
        self.assertEqual(feature_extractor.__class__.__name__    ,						"""NewFeatureExtractor"""		)
        self.assertTrue(feature_extractor.is_local		)

        # If remote code is disabled, we load the local one.
        lowerCAmelCase_			:      Dict			=   AutoFeatureExtractor.from_pretrained(
            """hf-internal-testing/test_dynamic_feature_extractor"""    ,						trust_remote_code=lowerCamelCase		)
        self.assertEqual(feature_extractor.__class__.__name__    ,						"""NewFeatureExtractor"""		)
        self.assertTrue(feature_extractor.is_local		)

        # If remote is enabled, we load from the Hub
        lowerCAmelCase_			:      List[Any]			=   AutoFeatureExtractor.from_pretrained(
            """hf-internal-testing/test_dynamic_feature_extractor"""    ,						trust_remote_code=lowerCamelCase		)
        self.assertEqual(feature_extractor.__class__.__name__    ,						"""NewFeatureExtractor"""		)
        self.assertTrue(not hasattr(lowerCamelCase    ,						"""is_local"""		)		)

      finally:
        if "custom" in CONFIG_MAPPING._extra_content:
          del CONFIG_MAPPING._extra_content["custom"]
        if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
          del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]







'''simple docstring'''






import re
import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin






class    __snake_case       (   _SCREAMING_SNAKE_CASE):







    """simple docstring"""



    lowercase    =					['image_processor', 'tokenizer']
    lowercase    =					'AutoImageProcessor'
    lowercase    =					'AutoTokenizer'

    def __init__(   self    :  int    ,						lowerCamelCase    :  List[str]=None    ,						lowerCamelCase    :  Union[str, Any]=None    ,						**lowerCamelCase    :  str		)   ->  Tuple:
      lowerCAmelCase_			:      str			=   None
      if "feature_extractor" in kwargs:
        warnings.warn(
            """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
            """ instead."""    ,						lowerCamelCase    ,						)
        lowerCAmelCase_			:      Tuple			=   kwargs.pop("""feature_extractor"""		)

      lowerCAmelCase_			:      Tuple			=   image_processor if image_processor is not None else feature_extractor
      if image_processor is None:
        raise ValueError("""You need to specify an `image_processor`."""		)
      if tokenizer is None:
        raise ValueError("""You need to specify a `tokenizer`."""		)

      super().__init__(lowerCamelCase    ,						lowerCamelCase		)
      lowerCAmelCase_			:      Optional[int]			=   self.image_processor
      lowerCAmelCase_			:      Any			=   False

    def __call__(   self    :  List[Any]    ,						*lowerCamelCase    :  str    ,						**lowerCamelCase    :  Tuple		)   ->  Union[str, Any]:
      # For backward compatibility
      if self._in_target_context_manager:
        return self.current_processor(*lowerCamelCase    ,						**lowerCamelCase		)

      lowerCAmelCase_			:      Any			=   kwargs.pop("""images"""    ,						lowerCamelCase		)
      lowerCAmelCase_			:      Dict			=   kwargs.pop("""text"""    ,						lowerCamelCase		)
      if len(lowerCamelCase		) > 0:
        lowerCAmelCase_			:      str			=   args[0]
        lowerCAmelCase_			:      Dict			=   args[1:]

      if images is None and text is None:
        raise ValueError("""You need to specify either an `images` or `text` input to process."""		)

      if images is not None:
        lowerCAmelCase_			:      Any			=   self.image_processor(lowerCamelCase    ,						*lowerCamelCase    ,						**lowerCamelCase		)
      if text is not None:
        lowerCAmelCase_			:      str			=   self.tokenizer(lowerCamelCase    ,						**lowerCamelCase		)

      if text is None:
        return inputs
      elif images is None:
        return encodings
      else:
        lowerCAmelCase_			:      Dict			=   encodings["""input_ids"""]
        return inputs

    def __lowercase   (   self    :  str    ,						*lowerCamelCase    :  Union[str, Any]    ,						**lowerCamelCase    :  Union[str, Any]		)   ->  List[str]:
      return self.tokenizer.batch_decode(*lowerCamelCase    ,						**lowerCamelCase		)

    def __lowercase   (   self    :  List[Any]    ,						*lowerCamelCase    :  Any    ,						**lowerCamelCase    :  Union[str, Any]		)   ->  List[str]:
      return self.tokenizer.decode(*lowerCamelCase    ,						**lowerCamelCase		)

    @contextmanager
    def __lowercase   (   self    :  str		)   ->  Union[str, Any]:
      warnings.warn(
          """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
          """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
          """your images inputs, or in a separate call."""		)
      lowerCAmelCase_			:      List[str]			=   True
      lowerCAmelCase_			:      Optional[Any]			=   self.tokenizer
      yield
      lowerCAmelCase_			:      List[Any]			=   self.image_processor
      lowerCAmelCase_			:      List[str]			=   False

    def __lowercase   (   self    :  Dict    ,						lowerCamelCase    :  List[str]    ,						lowerCamelCase    :  str=False    ,						lowerCamelCase    :  List[Any]=None		)   ->  Optional[int]:
      if added_vocab is None:
        lowerCAmelCase_			:      str			=   self.tokenizer.get_added_vocab()

      lowerCAmelCase_			:      Union[str, Any]			=   {}

      while tokens:
        lowerCAmelCase_			:      Dict			=   re.search(R"""<s_(.*?)>"""    ,						lowerCamelCase    ,						re.IGNORECASE		)
        if start_token is None:
          break
        lowerCAmelCase_			:      Tuple			=   start_token.group(1		)
        lowerCAmelCase_			:      Tuple			=   re.search(RF'</s_{key}>'    ,						lowerCamelCase    ,						re.IGNORECASE		)
        lowerCAmelCase_			:      Tuple			=   start_token.group()
        if end_token is None:
          lowerCAmelCase_			:      str			=   tokens.replace(lowerCamelCase    ,						""""""		)
        else:
          lowerCAmelCase_			:      List[str]			=   end_token.group()
          lowerCAmelCase_			:      Dict			=   re.escape(lowerCamelCase		)
          lowerCAmelCase_			:      int			=   re.escape(lowerCamelCase		)
          lowerCAmelCase_			:      Dict			=   re.search(F'{start_token_escaped}(.*?){end_token_escaped}'    ,						lowerCamelCase    ,						re.IGNORECASE		)
          if content is not None:
            lowerCAmelCase_			:      str			=   content.group(1		).strip()
            if r"<s_" in content and r"</s_" in content:  # non-leaf node
              lowerCAmelCase_			:      str			=   self.tokenajson(lowerCamelCase    ,						is_inner_value=lowerCamelCase    ,						added_vocab=lowerCamelCase		)
              if value:
                if len(lowerCamelCase		) == 1:
                  lowerCAmelCase_			:      List[Any]			=   value[0]
                lowerCAmelCase_			:      Optional[Any]			=   value
            else:  # leaf nodes
              lowerCAmelCase_			:      List[str]			=   []
              for leaf in content.split(R"""<sep/>"""		):
                lowerCAmelCase_			:      Union[str, Any]			=   leaf.strip()
                if leaf in added_vocab and leaf[0] == "<" and leaf[-2:] == "/>":
                  lowerCAmelCase_			:      Any			=   leaf[1:-2]  # for categorical special tokens
                output[key].append(lowerCamelCase		)
              if len(output[key]		) == 1:
                lowerCAmelCase_			:      Optional[Any]			=   output[key][0]

          lowerCAmelCase_			:      List[Any]			=   tokens[tokens.find(lowerCamelCase		) + len(lowerCamelCase		) :].strip()
          if tokens[:6] == r"<sep/>":  # non-leaf nodes
            return [output] + self.tokenajson(tokens[6:]    ,						is_inner_value=lowerCamelCase    ,						added_vocab=lowerCamelCase		)

      if len(lowerCamelCase		):
        return [output] if is_inner_value else output
      else:
        return [] if is_inner_value else {"text_sequence": tokens}

    @property
    def __lowercase   (   self    :  Dict		)   ->  int:
      warnings.warn(
          """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""    ,						lowerCamelCase    ,						)
      return self.image_processor_class





    @property
    def __lowercase   (   self    :  Dict		)   ->  Optional[Any]:
      warnings.warn(
          """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""    ,						lowerCamelCase    ,						)
      return self.image_processor



